from fastapi import FastAPI, Depends from fastapi.middleware.cors import CORSMiddleware from datetime import datetime from sqlalchemy.orm import Session from app.config import get_settings from app.database import engine, Base, get_db from app.routers import ( auth, dashboard, plants, production, inventory, suppliers, purchases, customers, sales, employees, finance, reports, alerts, users, treatments, branches, scan, weather, passports, farmers, farms, plant_activations, plant_support, farmer_analytics, accounting, cashbook, loans, production_costing, ) settings = get_settings() app = FastAPI( title="NurseryOps API", description="Nursery management system with QR tracing and AI disease detection", version="1.0.0", redirect_slashes=False, ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.include_router(auth.router, prefix=settings.api_prefix) app.include_router(dashboard.router, prefix=settings.api_prefix) app.include_router(plants.router, prefix=settings.api_prefix) app.include_router(production.router, prefix=settings.api_prefix) app.include_router(inventory.router, prefix=settings.api_prefix) app.include_router(suppliers.router, prefix=settings.api_prefix) app.include_router(purchases.router, prefix=settings.api_prefix) app.include_router(customers.router, prefix=settings.api_prefix) app.include_router(sales.router, prefix=settings.api_prefix) app.include_router(employees.router, prefix=settings.api_prefix) app.include_router(finance.router, prefix=settings.api_prefix) app.include_router(reports.router, prefix=settings.api_prefix) app.include_router(alerts.router, prefix=settings.api_prefix) app.include_router(users.router, prefix=settings.api_prefix) app.include_router(treatments.router, prefix=settings.api_prefix) app.include_router(branches.router, prefix=settings.api_prefix) app.include_router(scan.router, prefix=settings.api_prefix) app.include_router(weather.router, prefix=settings.api_prefix) app.include_router(passports.router, prefix=settings.api_prefix) app.include_router(farmers.router, prefix=settings.api_prefix) app.include_router(farms.router, prefix=settings.api_prefix) app.include_router(plant_activations.router, prefix=settings.api_prefix) app.include_router(plant_support.router, prefix=settings.api_prefix) app.include_router(farmer_analytics.router, prefix=settings.api_prefix) app.include_router(accounting.router, prefix=settings.api_prefix) app.include_router(cashbook.router, prefix=settings.api_prefix) app.include_router(loans.router, prefix=settings.api_prefix) app.include_router(production_costing.router, prefix=settings.api_prefix) @app.get("/s/{code}") def public_scan_page(code: str, db: Session = Depends(get_db)): from app.routers.scan import _get_scan_data, _scan_html from fastapi.responses import HTMLResponse 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) @app.get("/p/{passport_id}") def public_passport_page(passport_id: str, db: Session = Depends(get_db)): from app.routers.passports import _build_passport_from_batch from fastapi.responses import HTMLResponse from app.models.passport import PlantPassport, PassportScanLog passport = db.query(PlantPassport).filter( (PlantPassport.passport_id == passport_id) | (PlantPassport.id == passport_id) ).first() if not passport: return HTMLResponse( content="Not Found

🔍

Passport Not Found

The passport " + passport_id + " does not exist.

", status_code=404, ) scan_log = PassportScanLog( passport_id=passport.id, scan_source="qr_code", scanned_at=datetime.now(), ) db.add(scan_log) db.commit() return HTMLResponse(content=_passport_public_html(passport), status_code=200) def _passport_public_html(p) -> str: stage_colors = { "sowing": ("#f59e0b", "#fef3c7"), "germination": ("#3b82f6", "#dbeafe"), "growing": ("#22c55e", "#dcfce7"), "ready": ("#a855f7", "#f3e8ff"), "sold": ("#6b7280", "#f3f4f6"), } s_color, s_bg = stage_colors.get(p.current_growth_stage or "sowing", ("#6b7280", "#f3f4f6")) stage_label = (p.current_growth_stage or "sowing").title() return f''' {p.passport_id} - Plant Digital Passport

Plant Digital Passport

{p.passport_id}

{stage_label}

{p.plant_name or "N/A"}

Plant

{p.current_stock or 0}

Stock

🌱 Plant Info

{p.plant_name or "N/A"}

{p.scientific_name or ""}

{f'

Variety: {p.variety_name}

' if p.variety_name else ""} {f'

Category: {p.category_name}

' if p.category_name else ""} {f'

Batch: {p.batch_number}

' if p.batch_number else ""}

📦 Seed Source

{p.seed_lot_number or "N/A"}

Supplier: {p.seed_supplier or "N/A"}

Care Instructions

💧

Watering

Regular schedule

☀️

Sunlight

Full sun / Partial shade

🧪

Fertilizer

Monthly application

🐛

Pest Control

Regular inspection

Growth Timeline

{f'

Sowing Date

{p.sowing_date}

' if p.sowing_date else ""} {f'

Germination

{p.germination_date} ({p.germination_rate}%)

' if p.germination_date else ""} {f'

Transplant

{p.transplant_date}

' if p.transplant_date else ""}

Generated by NurseryOps

Passport: {p.passport_id}

''' @app.on_event("startup") def startup(): Base.metadata.create_all(bind=engine) _seed_defaults() def _clear_non_user_data(): from app.database import SessionLocal from sqlalchemy import text db = SessionLocal() try: db.execute(text("SET session_replication_role = 'replica'")) tables_to_clear = [ "passport_customer_views", "passport_scan_logs", "passport_photos", "passport_timeline_events", "plant_passports", "treatment_schedules", "disease_pest_records", "health_records", "treatment_details", "treatments", "transplant_records", "germination_records", "production_batches", "seed_lots", "plant_varieties", "plants", "plant_categories", "inventory_transactions", "inventory", "purchase_items", "purchase_orders", "purchase_requests", "grn", "sale_items", "sales", "quotation_items", "quotations", "customers", "suppliers", "attendance", "leaves", "employees", "journal_entries", "payments", "income", "expenses", "accounts", "notifications", "alerts", "audit_logs", "stock_transfers", "warehouses", ] for table in tables_to_clear: try: db.execute(text(f"DELETE FROM {table}")) except Exception: pass db.execute(text("SET session_replication_role = 'origin'")) db.commit() except Exception: db.rollback() finally: db.close() def _seed_branches(db): from app.models.branch import Branch extra = [ ("BR-002", "Noymile Branch", "Noymile"), ("BR-003", "Mohasthan Branch", "Mohasthan"), ("BR-004", "Seed & Agro Input Shop", "Agro Input Shop"), ] for code, name, address in extra: if not db.query(Branch).filter(Branch.code == code).first(): db.add(Branch(code=code, name=name, address=address, is_active=True)) def _seed_defaults(): from app.database import SessionLocal from app.models.user import Role, Permission, UserRole, RolePermission, User from app.models.branch import Branch db = SessionLocal() try: if db.query(Role).count() == 0: modules = ["plants", "production", "inventory", "sales", "purchases", "employees", "finance", "treatments", "reports", "users", "branches", "farmers", "farms", "farmer_plants", "plant_growth", "plant_treatments", "plant_health", "plant_harvest", "plant_issues", "plant_warranty", "farmer_support", "accounting"] actions = ["view", "create", "edit", "delete"] perms = {} for m in modules: for a in actions: p = Permission(module=m, action=a) db.add(p) db.flush() perms[(m, a)] = p admin_role = Role(name="super_admin", description="Full system access") db.add(admin_role) db.flush() for p in perms.values(): db.add(RolePermission(role_id=admin_role.id, permission_id=p.id)) manager_role = Role(name="manager", description="Branch manager") db.add(manager_role) db.flush() for m in modules: for a in actions: db.add(RolePermission(role_id=manager_role.id, permission_id=perms[(m, a)].id)) staff_role = Role(name="staff", description="General staff") db.add(staff_role) db.flush() for m in modules: db.add(RolePermission(role_id=staff_role.id, permission_id=perms[(m, "view")].id)) viewer_role = Role(name="viewer", description="Read-only access") db.add(viewer_role) db.flush() for m in modules: db.add(RolePermission(role_id=viewer_role.id, permission_id=perms[(m, "view")].id)) farmer_role = Role(name="farmer", description="Farmer self-service access") db.add(farmer_role) db.flush() farmer_modules = ["farmer_plants", "plant_growth", "plant_treatments", "plant_health", "plant_harvest", "plant_issues", "farmer_support"] for m in farmer_modules: db.add(RolePermission(role_id=farmer_role.id, permission_id=perms[(m, "view")].id)) db.add(RolePermission(role_id=farmer_role.id, permission_id=perms[(m, "create")].id)) db.flush() if db.query(Branch).count() == 0: main_branch = Branch(code="BR-001", name="Main Branch", address="Head Office", is_active=True) db.add(main_branch) _seed_branches(db) db.flush() from app.services.accounting import seed_chart_of_accounts seed_chart_of_accounts(db) if db.query(User).count() == 0: admin_role = db.query(Role).filter(Role.name == "super_admin").first() from app.services.auth import hash_password admin = User( username="admin", email="admin@nursery.com", password_hash=hash_password("admin123"), full_name="System Administrator", is_active=True, ) if admin_role: admin.roles = [admin_role] db.add(admin) db.commit() except Exception: db.rollback() finally: db.close() @app.get("/") def root(): return {"message": "NurseryOps API is running"}