from __future__ import annotations import uuid import io import base64 import socket from datetime import datetime, date from fastapi import APIRouter, Depends, HTTPException, Response from fastapi.responses import HTMLResponse from sqlalchemy.orm import Session from pydantic import BaseModel from typing import Optional, List from app.database import get_db from app.models.production import ProductionBatch, SeedLot, GerminationRecord, TransplantRecord from app.models.plant import Plant, PlantVariety, PlantCategory from app.models.supplier import Supplier from app.models.employee import Employee from app.models.treatment import Treatment, TreatmentDetail, HealthRecord, DiseasePestRecord from app.services.auth import get_current_user router = APIRouter(tags=["scan"]) def _get_server_host() -> str: try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except Exception: return "localhost" def _get_scan_data(code: str, db: Session) -> dict: batch = db.query(ProductionBatch).filter( (ProductionBatch.id == code) | (ProductionBatch.batch_number == code) ).first() if not batch: return None plant = db.query(Plant).filter(Plant.id == batch.plant_id).first() if batch.plant_id else None variety = db.query(PlantVariety).filter(PlantVariety.id == plant.variety_id).first() if plant and plant.variety_id else None category = db.query(PlantCategory).filter(PlantCategory.id == plant.category_id).first() if plant and plant.category_id else None seed_lot = db.query(SeedLot).filter(SeedLot.id == batch.seed_lot_id).first() if batch.seed_lot_id else None supplier = db.query(Supplier).filter(Supplier.id == seed_lot.supplier_id).first() if seed_lot and seed_lot.supplier_id else None employee = db.query(Employee).filter(Employee.id == batch.employee_id).first() if batch.employee_id else None germination = db.query(GerminationRecord).filter(GerminationRecord.batch_id == batch.id).order_by(GerminationRecord.date).all() transplants = db.query(TransplantRecord).filter(TransplantRecord.batch_id == batch.id).order_by(TransplantRecord.date).all() treatments = db.query(Treatment).filter(Treatment.batch_id == batch.id).order_by(Treatment.treatment_date).all() health = db.query(HealthRecord).filter(HealthRecord.batch_id == batch.id).order_by(HealthRecord.record_date).all() diseases = db.query(DiseasePestRecord).filter(DiseasePestRecord.batch_id == batch.id).order_by(DiseasePestRecord.record_date).all() timeline = [] timeline.append({"date": batch.sowing_date.isoformat(), "event": "Sowing", "details": f"{batch.sown_quantity} seeds planted", "icon": "seed"}) if seed_lot: timeline.append({"date": (seed_lot.purchase_date or batch.sowing_date).isoformat(), "event": "Seed Purchase", "details": f"Lot: {seed_lot.lot_number}", "icon": "purchase"}) for g in germination: timeline.append({"date": g.date.isoformat(), "event": "Germination Check", "details": f"{g.germinated_quantity} germinated ({g.success_rate}%)", "icon": "germination"}) for t in transplants: timeline.append({"date": t.date.isoformat(), "event": "Transplant", "details": f"{t.from_container} → {t.to_container} ({t.quantity})", "icon": "transplant"}) for tr in treatments: detail_data = None detail_obj = db.query(TreatmentDetail).filter(TreatmentDetail.treatment_id == tr.id).first() if detail_obj: detail_data = {k: v for k, v in detail_obj.__dict__.items() if v is not None and k != '_sa_instance_state' and k != 'id' and k != 'treatment_id'} timeline.append({ "date": tr.treatment_date.isoformat(), "event": tr.treatment_type.replace('_', ' ').title(), "details": tr.notes or str(detail_data) if detail_data else "", "icon": "treatment", }) for h in health: timeline.append({"date": h.record_date.isoformat(), "event": "Health Check", "details": f"Score: {h.health_score}/100" if h.health_score else "", "icon": "health"}) for d in diseases: timeline.append({"date": d.record_date.isoformat(), "event": f"{d.type.title()}: {d.name}", "details": f"Severity: {d.severity} ({d.status})", "icon": "disease"}) timeline.sort(key=lambda x: x["date"]) return { "batch_id": batch.id, "batch_number": batch.batch_number, "status": batch.status, "sowing_date": batch.sowing_date.isoformat(), "expected_ready_date": batch.expected_ready_date.isoformat() if batch.expected_ready_date else None, "sown_quantity": batch.sown_quantity, "current_quantity": batch.current_quantity, "tray_number": batch.tray_number, "location": batch.location, "plant_name": plant.name if plant else None, "plant_scientific": plant.scientific_name if plant else None, "variety_name": variety.name if variety else None, "category_name": category.name if category else None, "seed_lot_number": seed_lot.lot_number if seed_lot else None, "seed_supplier": supplier.name if seed_lot and seed_lot.supplier_id else None, "employee_name": employee.name if employee else None, "germination_records": [{"date": g.date.isoformat(), "quantity": g.germinated_quantity, "rate": g.success_rate} for g in germination], "transplant_records": [{"date": t.date.isoformat(), "from": t.from_container, "to": t.to_container, "quantity": t.quantity} for t in transplants], "treatments": [{"number": t.treatment_number, "type": t.treatment_type, "date": t.treatment_date.isoformat(), "status": t.status} for t in treatments], "health_records": [{"date": h.record_date.isoformat(), "score": h.health_score, "height": h.plant_height_cm, "mortality": h.mortality_rate} for h in health], "disease_pest_records": [{"date": d.record_date.isoformat(), "name": d.name, "type": d.type, "severity": d.severity, "status": d.status} for d in diseases], "timeline": timeline, } def _generate_qr_base64(data: str) -> str: import qrcode qr = qrcode.QRCode(version=1, error_correction=qrcode.constants.ERROR_CORRECT_H, box_size=8, border=2) qr.add_data(data) qr.make(fit=True) img = qr.make_image(fill_color="#166534", back_color="white") buf = io.BytesIO() img.save(buf, format="PNG") return base64.b64encode(buf.getvalue()).decode() def _generate_qr_svg(data: str) -> str: import qrcode import qrcode.image.svg qr = qrcode.QRCode(version=1, error_correction=qrcode.constants.ERROR_CORRECT_M, box_size=10, border=2) qr.add_data(data) qr.make(fit=True) factory = qrcode.image.svg.SvgPathImage img = qr.make_image(image_factory=factory) buf = io.BytesIO() img.save(buf) return buf.getvalue().decode() def _scan_html(data: dict) -> str: status_colors = { "sowing": ("#f59e0b", "Sowing"), "germination": ("#3b82f6", "Germination"), "growing": ("#22c55e", "Growing"), "ready": ("#a855f7", "Ready"), "sold": ("#6b7280", "Sold"), } s_color, s_label = status_colors.get(data["status"], ("#6b7280", data["status"])) survival = round((data["current_quantity"] / data["sown_quantity"]) * 100, 1) if data["sown_quantity"] > 0 else 0 germination_rate = None if data["germination_records"]: germination_rate = data["germination_records"][-1]["rate"] timeline_html = "" icon_map = { "seed": ("#f59e0b", "#fef3c7"), "purchase": ("#3b82f6", "#dbeafe"), "germination": ("#22c55e", "#dcfce7"), "transplant": ("#a855f7", "#f3e8ff"), "treatment": ("#14b8a6", "#ccfbf1"), "health": ("#10b981", "#d1fae5"), "disease": ("#ef4444", "#fee2e2"), } for item in data["timeline"]: ic, bg = icon_map.get(item["icon"], ("#6b7280", "#f3f4f6")) timeline_html += f'''
{item["event"]} {item["date"]}
{f'

{item["details"]}

' if item.get("details") else ""}
''' health_html = "" if data["health_records"]: cards = "" for h in data["health_records"]: score = h.get("score") score_color = "#22c55e" if score and score >= 80 else "#f59e0b" if score and score >= 50 else "#ef4444" cards += f'''
{h["date"]}
{f'{score}/100' if score else ""} {f'{h["height"]}cm' if h.get("height") else ""}
''' health_html = f'''

♥ Health Records

{cards}
''' disease_html = "" if data["disease_pest_records"]: cards = "" for d in data["disease_pest_records"]: sev_bg = "#fee2e2" if d["severity"] == "high" else "#ffedd5" if d["severity"] == "medium" else "#fef9c3" sev_color = "#dc2626" if d["severity"] == "high" else "#ea580c" if d["severity"] == "medium" else "#ca8a04" status_bg = "#d1fae5" if d["status"] == "resolved" else "#fee2e2" status_color = "#059669" if d["status"] == "resolved" else "#dc2626" cards += f'''
{d["name"]} ({d["type"]})
{d["severity"]} {d["status"]}
''' disease_html = f'''

⚠ Disease & Pest Log

{cards}
''' return f''' {data["batch_number"]} - Traceability Report

Seedling Traceability Report

{data["batch_number"]}

{s_label}
Growth Progress{survival}%

{data["sown_quantity"]}

Sown

{data["current_quantity"]}

Current

{survival}%

Survival

🌱 Plant

{data["plant_name"] or "N/A"}

{data["plant_scientific"] or ""}

{f'

Variety: {data["variety_name"]}

' if data.get("variety_name") else ""} {f'

Category: {data["category_name"]}

' if data.get("category_name") else ""}

卡车 Seed Source

{data["seed_lot_number"] or "N/A"}

Supplier: {data["seed_supplier"] or "N/A"}

⏱ Complete Timeline

{timeline_html}
{health_html} {disease_html}

{len(data["treatments"])}

Treatments

{germination_rate if germination_rate is not None else "N/A"}{"%" if germination_rate is not None else ""}

Germination

{len(data["transplant_records"])}

Transplants

Generated by NurseryOps

Batch: {data["batch_number"]} · {data["sowing_date"]}

''' @router.get("/scan/{code}") def scan_batch(code: str, db: Session = Depends(get_db)): data = _get_scan_data(code, db) if not data: raise HTTPException(status_code=404, detail="Batch not found") return data @router.get("/scan-page/{code}", response_class=HTMLResponse) def scan_page(code: str, db: Session = Depends(get_db)): data = _get_scan_data(code, db) if not data: return HTMLResponse( content="Not Found

🔍

Batch Not Found

The batch code {code} does not exist.

", status_code=404, ) return HTMLResponse(content=_scan_html(data), status_code=200) @router.get("/qr/{batch_id}") def get_batch_qr(batch_id: str, request_url: str = None, db: Session = Depends(get_db)): batch = db.query(ProductionBatch).filter( (ProductionBatch.id == batch_id) | (ProductionBatch.batch_number == batch_id) ).first() if not batch: raise HTTPException(status_code=404, detail="Batch not found") host = _get_server_host() scan_url = f"http://{host}:8000/s/{batch.batch_number}" qr_b64 = _generate_qr_base64(scan_url) return { "batch_number": batch.batch_number, "qr_base64": qr_b64, "scan_url": scan_url, } @router.get("/qr/batch/{batch_id}/print") def get_batch_qr_print(batch_id: str, db: Session = Depends(get_db)): batch = db.query(ProductionBatch).filter(ProductionBatch.id == batch_id).first() if not batch: raise HTTPException(status_code=404, detail="Batch not found") plant = db.query(Plant).filter(Plant.id == batch.plant_id).first() if batch.plant_id else None host = _get_server_host() scan_url = f"http://{host}:8000/s/{batch.batch_number}" qr_b64 = _generate_qr_base64(scan_url) return { "batch_number": batch.batch_number, "plant_name": plant.name if plant else "N/A", "sowing_date": batch.sowing_date.isoformat(), "status": batch.status, "quantity": batch.sown_quantity, "qr_base64": qr_b64, "scan_url": scan_url, }