from flask import Flask, request, jsonify from flask_cors import CORS import json from datetime import datetime from typing import Optional, Dict, Any, List import re import os import io from pypdf import PdfReader from docx import Document from firestore_client import get_firestore_client from gemini_client import ask_gpt from prompt_instructions import build_system_message from role_access import get_allowed_collections from data_fetcher import fetch_data_from_firestore from data_planner import determine_data_requirements from resolver import resolve_user_context from schema_utils import has_field, resolve_field app = Flask(__name__) CORS(app) db = get_firestore_client() # -- helpers -------------------------------------------------------------- def to_jsonable(obj): """Recursively convert Firestore types (e.g., DatetimeWithNanoseconds) to JSON-safe values.""" # Firestore timestamp types come through as DatetimeWithNanoseconds (duck-type datetime) if isinstance(obj, datetime): return obj.isoformat() # Some Firestore SDKs expose Timestamp-like objects without subclassing datetime. # Fallback: detect presence of isoformat() if hasattr(obj, "isoformat") and callable(getattr(obj, "isoformat")): try: return obj.isoformat() except Exception: pass if isinstance(obj, dict): return {k: to_jsonable(v) for k, v in obj.items()} if isinstance(obj, list): return [to_jsonable(v) for v in obj] return obj # 🔧 Normalize Gemini plan into proper Firestore fetch format def normalize_plan(plan: dict, token_map: Optional[Dict[str, Any]] = None) -> dict: token_map = token_map or {} filters = plan.get("filters", {}) or {} planned_cols = plan.get("collections", []) or [] def canonical_value(key, val): # replace tokens like {{participantId}} if isinstance(val, str) and val in token_map: val = token_map[val] # special-case status normalization if key == "status" and val == "running": return "active" return val collections_out = [] for col in planned_cols: # Allow both strings ("participants") and objects ({"name":"participants","fields":[...]}). name = col["name"] if isinstance(col, dict) else col col_filters = [] for k, v in filters.items(): canon_key = resolve_field(name, k) # only add filters that exist for this collection (schema-aware) if has_field(name, canon_key): col_filters.append({ "field": canon_key, "op": "==", "value": canonical_value(k, v), }) # else: skip invalid field for this collection collections_out.append({"name": name, "filters": col_filters, "limit": 50}) return {"collections": collections_out} def _norm(v): return str(v or "").strip().lower() def _safe_number(v, default=0): try: n = float(v) return n except Exception: return default def _clamp_pct(n): try: return max(0, min(100, round(float(n)))) except Exception: return 0 def _normalize_metric(metric: Optional[str]) -> Optional[str]: m = _norm(metric) if not m: return None if "hour" in m: return "hours" if "session" in m: return "sessions" if "document" in m: return "documents" if "deliverable" in m: return "deliverables" if "report" in m: return "reports" if "submission" in m: return "submissions" return None def _proof_rules_for_metric(metric_key: Optional[str]) -> List[Dict[str, Any]]: if metric_key == "sessions": return [ { "type": "attendance", "label": "Attendance register / session proof", "reason": "Sessions usually require attendance or meeting evidence.", "required": True, }, { "type": "minutes", "label": "Session notes / minutes", "reason": "Session outcomes should be supported by notes or minutes.", "required": False, }, ] if metric_key in ["reports", "documents", "deliverables", "submissions"]: return [ { "type": "document", "label": "Output file / submitted document", "reason": "This metric usually requires the actual file or submission proof.", "required": True, } ] if metric_key == "hours": return [ { "type": "supporting_evidence", "label": "Supporting evidence", "reason": "Hours-only updates should still have some supporting evidence where possible.", "required": False, } ] return [] def _build_intervention_analysis_prompt(payload: Dict[str, Any]) -> str: return f""" You are analysing a consultant intervention progress update. Return STRICT JSON only with this shape: {{ "summary": "short summary", "polishedNotes": "professional progress notes", "activities": ["..."], "outputs": ["..."], "blockers": ["..."], "nextSteps": ["..."], "suggestedHours": 0, "suggestedUnitsCompleted": 0, "confidence": 0, "completionSignal": "none|partial|strong", "warnings": ["..."], "proofHints": [ {{ "type": "attendance|minutes|document|photo|supporting_evidence|submission_receipt|other", "label": "short label", "reason": "why this is needed" }} ] }} Rules: - Use the user's notes/transcript/message as the primary source. - Do not invent facts not grounded in the input. - suggestedHours should be conservative. - suggestedUnitsCompleted should only be given if the described work clearly maps to a measurable unit. - completionSignal means: - "none" = no sign of completion - "partial" = meaningful progress but not clearly complete - "strong" = strong signal that deliverable/work may be complete - warnings should mention ambiguity, weak evidence, or missing detail. - polishedNotes should be concise and professional. Intervention context: {json.dumps(payload, ensure_ascii=False)} """.strip() def _extract_json_block(text: str) -> Dict[str, Any]: text = (text or "").strip() try: return json.loads(text) except Exception: pass match = re.search(r"\{.*\}", text, flags=re.DOTALL) if match: try: return json.loads(match.group(0)) except Exception: pass return { "summary": "", "polishedNotes": text, "activities": [], "outputs": [], "blockers": [], "nextSteps": [], "suggestedHours": None, "suggestedUnitsCompleted": None, "confidence": 40, "completionSignal": "none", "warnings": ["AI response was not clean JSON. Parsed fallback response."], "proofHints": [], } def _merge_proof_hints(metric_rules: List[Dict[str, Any]], ai_hints: List[Dict[str, Any]]) -> List[Dict[str, Any]]: merged = [] seen = set() for item in metric_rules + (ai_hints or []): t = _norm(item.get("type")) label = str(item.get("label") or "").strip() key = f"{t}|{label.lower()}" if not t and not label: continue if key in seen: continue seen.add(key) merged.append({ "type": t or "other", "label": label or "Supporting evidence", "reason": str(item.get("reason") or "").strip(), "required": bool(item.get("required", False)), }) return merged def _calculate_progress_suggestion(intervention: Dict[str, Any], ai_result: Dict[str, Any]) -> Dict[str, Any]: target_type = _norm(intervention.get("targetType")) target_value = _safe_number(intervention.get("targetValue"), 0) current_progress = _clamp_pct(_safe_number(intervention.get("progress"), 0)) current_actual = _safe_number(intervention.get("targetActual"), 0) metric_key = _normalize_metric(intervention.get("targetMetric")) suggested_hours = ai_result.get("suggestedHours") try: suggested_hours = float(suggested_hours) if suggested_hours is not None else None except Exception: suggested_hours = None suggested_units = ai_result.get("suggestedUnitsCompleted") try: suggested_units = float(suggested_units) if suggested_units is not None else None except Exception: suggested_units = None if target_type == "percentage": # For percentage interventions, AI can only suggest a conservative bump. base_bump = 0 signal = _norm(ai_result.get("completionSignal")) if signal == "strong": base_bump = 20 elif signal == "partial": base_bump = 10 elif suggested_hours and suggested_hours > 0: base_bump = 5 suggested_progress = _clamp_pct(current_progress + base_bump) return { "targetMode": "percentage", "metricKey": metric_key, "suggestedHours": suggested_hours, "suggestedUnitsCompleted": None, "currentActual": None, "suggestedActualAfter": None, "suggestedProgressPct": suggested_progress, "overTargetBy": 0, } if target_type == "number" and target_value > 0: units = suggested_units if metric_key == "hours" and (units is None or units <= 0): units = suggested_hours if units is None or units <= 0: return { "targetMode": "number", "metricKey": metric_key, "suggestedHours": suggested_hours, "suggestedUnitsCompleted": None, "currentActual": current_actual, "suggestedActualAfter": current_actual, "suggestedProgressPct": current_progress, "overTargetBy": 0, } actual_after = current_actual + units progress_after = _clamp_pct((actual_after / target_value) * 100) over_target_by = max(0, actual_after - target_value) return { "targetMode": "number", "metricKey": metric_key, "suggestedHours": suggested_hours, "suggestedUnitsCompleted": units, "currentActual": current_actual, "suggestedActualAfter": actual_after, "suggestedProgressPct": progress_after, "overTargetBy": over_target_by, } return { "targetMode": "none", "metricKey": metric_key, "suggestedHours": suggested_hours, "suggestedUnitsCompleted": suggested_units, "currentActual": None, "suggestedActualAfter": None, "suggestedProgressPct": current_progress, "overTargetBy": 0, } ALLOWED_COURSE_SOURCE_EXTENSIONS = {"pdf", "docx"} MAX_SOURCE_TEXT_CHARS = 60000 def _allowed_course_source(filename: str) -> bool: if not filename or "." not in filename: return False return filename.rsplit(".", 1)[1].lower() in ALLOWED_COURSE_SOURCE_EXTENSIONS def _clean_extracted_text(text: str) -> str: if not text: return "" text = text.replace("\x00", " ") text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\n{3,}", "\n\n", text) return text.strip() def _extract_text_from_pdf_bytes(file_bytes: bytes) -> str: reader = PdfReader(io.BytesIO(file_bytes)) pages = [] for page in reader.pages: try: pages.append(page.extract_text() or "") except Exception: pages.append("") return _clean_extracted_text("\n\n".join(pages)) def _extract_text_from_docx_bytes(file_bytes: bytes) -> str: doc = Document(io.BytesIO(file_bytes)) lines: List[str] = [] for p in doc.paragraphs: txt = (p.text or "").strip() if txt: lines.append(txt) for table in doc.tables: for row in table.rows: row_text = " | ".join((cell.text or "").strip() for cell in row.cells if (cell.text or "").strip()) if row_text: lines.append(row_text) return _clean_extracted_text("\n".join(lines)) def _extract_course_source_text(filename: str, file_bytes: bytes) -> str: ext = filename.rsplit(".", 1)[1].lower() if ext == "pdf": return _extract_text_from_pdf_bytes(file_bytes) if ext == "docx": return _extract_text_from_docx_bytes(file_bytes) raise ValueError("Unsupported file type") def _truncate_source_text(text: str, limit: int = MAX_SOURCE_TEXT_CHARS) -> str: if len(text) <= limit: return text return text[:limit] def _build_course_outline_prompt(source_text: str, filename: str) -> str: return f""" You are designing a practical learning course outline from source material. Return STRICT JSON only with this exact shape: {{ "courseTitle": "string", "courseDescription": "string", "difficulty": "beginner|intermediate|advanced", "category": "string", "courseType": "string", "estimatedTotalDuration": "string", "learningObjectives": ["string"], "modules": [ {{ "type": "lesson|quiz|assignment|review", "title": "string", "description": "string", "duration": "e.g. 20m or 1h", "content": "only for lesson when useful", "assignmentPrompt": "only for assignment when useful", "answerKey": "only for assignment when useful", "questions": [ {{ "question": "string", "options": ["string", "string", "string", "string"], "correctAnswer": 0 }} ] }} ], "warnings": ["string"] }} Rules: - Build a course outline grounded in the uploaded document. - Prefer 4 to 12 modules unless the source strongly suggests otherwise. - Most modules should be lessons. - Include quizzes only where knowledge checks make sense. - Include assignments only when there is something practical to apply. - Include review modules only when useful for recap. - Every lesson must have a realistic duration estimate. - estimatedTotalDuration must reflect the sum of lesson durations approximately. - Keep titles practical and clean. - Do not invent niche facts that are not supported by the source. - If the document is too thin, still produce a usable outline and add a warning. - If the content looks like a scanned PDF with poor extraction, say so in warnings. Filename: {filename} Source document text: {source_text} """.strip() def _normalize_outline_json(ai_result: Dict[str, Any]) -> Dict[str, Any]: raw_modules = ai_result.get("modules") or [] out_modules = [] for idx, mod in enumerate(raw_modules): mtype = str(mod.get("type") or "lesson").strip().lower() if mtype not in ["lesson", "quiz", "assignment", "review"]: mtype = "lesson" base = { "id": f"module-{idx + 1}", "type": mtype, "title": str(mod.get("title") or f"Module {idx + 1}").strip(), "description": str(mod.get("description") or "").strip(), } if mtype == "lesson": base["duration"] = str(mod.get("duration") or "20m").strip() base["content"] = str(mod.get("content") or "").strip() base["videoUrls"] = [] base["imageUrls"] = [] elif mtype == "quiz": questions = [] for qidx, q in enumerate(mod.get("questions") or []): options = q.get("options") or [] while len(options) < 4: options.append("") questions.append({ "id": f"q-{idx + 1}-{qidx + 1}", "question": str(q.get("question") or "").strip(), "options": [str(x or "").strip() for x in options[:4]], "correctAnswer": int(q.get("correctAnswer") or 0), }) base["questions"] = questions elif mtype == "assignment": base["assignmentPrompt"] = str(mod.get("assignmentPrompt") or "").strip() base["answerKey"] = str(mod.get("answerKey") or "").strip() out_modules.append(base) return { "courseTitle": str(ai_result.get("courseTitle") or "").strip(), "courseDescription": str(ai_result.get("courseDescription") or "").strip(), "difficulty": str(ai_result.get("difficulty") or "beginner").strip().lower(), "category": str(ai_result.get("category") or "General").strip(), "courseType": str(ai_result.get("courseType") or "Foundational").strip(), "estimatedTotalDuration": str(ai_result.get("estimatedTotalDuration") or "").strip(), "learningObjectives": [ str(x).strip() for x in (ai_result.get("learningObjectives") or []) if str(x).strip() ], "modules": out_modules, "warnings": [ str(x).strip() for x in (ai_result.get("warnings") or []) if str(x).strip() ], } # --- SME Intake helpers ------------------------------------------------- ALLOWED_SME_DOC_EXTENSIONS = {"pdf", "docx", "txt", "png", "jpg", "jpeg"} MAX_SME_DOC_TEXT_CHARS = 45000 def _allowed_sme_doc(filename: str) -> bool: if not filename or "." not in filename: return False return filename.rsplit(".", 1)[1].lower() in ALLOWED_SME_DOC_EXTENSIONS def _extract_text_from_txt_bytes(file_bytes: bytes) -> str: try: return _clean_extracted_text(file_bytes.decode("utf-8", errors="ignore")) except Exception: return "" def _extract_sme_doc_text(filename: str, file_bytes: bytes) -> str: ext = filename.rsplit(".", 1)[1].lower() if ext == "pdf": return _extract_text_from_pdf_bytes(file_bytes) if ext == "docx": return _extract_text_from_docx_bytes(file_bytes) if ext == "txt": return _extract_text_from_txt_bytes(file_bytes) # For images, we are not doing OCR here. # Send metadata to AI, but text extraction stays blank. if ext in ["png", "jpg", "jpeg"]: return "" return "" def _fetch_active_programs(company_code: Optional[str] = None) -> List[Dict[str, Any]]: try: ref = db.collection("programs") if company_code: docs = ref.where("companyCode", "==", company_code).stream() else: docs = ref.stream() programs = [] for doc_snap in docs: data = doc_snap.to_dict() or {} status = _norm(data.get("status") or data.get("programStatus")) if status and status not in ["active", "open", "running"]: continue eligibility = data.get("eligibilityCriteria") or {} programs.append({ "id": doc_snap.id, "companyCode": data.get("companyCode") or "", "name": data.get("name") or "", "description": data.get("description") or "", "type": data.get("type") or "", "cohortYear": data.get("cohortYear") or "", "startDate": to_jsonable(data.get("startDate")), "endDate": to_jsonable(data.get("endDate")), "maxCapacity": data.get("maxCapacity"), "registrationLink": data.get("registrationLink") or "", "eligibilityCriteria": eligibility, "eligibilitySummary": { "minimumAge": eligibility.get("minimumAge"), "maximumAge": eligibility.get("maximumAge"), "gender": eligibility.get("gender"), "sector": eligibility.get("sector"), "province": eligibility.get("province"), "minYearsOfTrading": eligibility.get("minYearsOfTrading"), "bbbeeLevel": eligibility.get("bbbeeLevel"), "minYouthOwnershipPercent": eligibility.get("minYouthOwnershipPercent"), "minBlackOwnershipPercent": eligibility.get("minBlackOwnershipPercent"), "minFemaleOwnershipPercent": eligibility.get("minFemaleOwnershipPercent"), }, }) return programs[:50] except Exception as e: print("fetch_active_programs_failed:", e) return [] def _fetch_intervention_catalog(company_code: str) -> List[Dict[str, Any]]: try: docs = ( db.collection("interventions") .where("companyCode", "==", company_code) .stream() ) interventions = [] for doc_snap in docs: data = doc_snap.to_dict() or {} status = _norm(data.get("status")) if status and status not in ["active", "enabled", "approved"]: continue interventions.append({ "id": doc_snap.id, "title": data.get("title") or data.get("interventionTitle") or data.get("name") or "", "department": data.get("departmentName") or data.get("areaOfSupport") or data.get("department") or "", "description": data.get("description") or data.get("objective") or "", "tags": data.get("tags") or [], "stage": data.get("stage") or data.get("businessStage") or "", }) return interventions[:80] except Exception as e: print("fetch_intervention_catalog_failed:", e) return [] def _build_sme_intake_prompt(payload: Dict[str, Any]) -> str: return f""" You are analysing an unregistered SME intake for an incubation/support platform. Return STRICT JSON only with this exact shape: {{ "detectedLanguages": ["English"], "confidence": 0, "extractedProfile": {{ "businessName": "", "beneficiaryName": "", "contactPerson": "", "email": "", "phone": "", "registrationNumber": "", "taxNumber": "", "sector": "", "province": "", "city": "", "yearsOperating": "", "employeeCount": "", "monthlyRevenue": "", "mainChallenges": [] }}, "extractedApplication": {{ "motivation": "", "challenges": "", "businessDescription": "", "supportNeeded": [], "documentsMentioned": [] }}, "documentFindings": [ {{ "documentType": "id_document|cipc|tax_pin|other", "filename": "", "extractedFields": {{}}, "confidence": 0, "warnings": [] }} ], "missingFields": [ {{ "field": "", "label": "", "question": "", "reason": "" }} ], "missingDocuments": [ {{ "type": "id_document|cipc|tax_pin|other", "reason": "" }} ], "nextQuestion": "", "decision": {{ "stage": "idea|startup|early_growth|struggling|scaling|market_ready", "urgencyLevel": "low|medium|high|urgent", "urgencyScore": 0, "riskLevel": "low|medium|high", "recommendedProgramId": "", "recommendedProgramName": "", "recommendedInterventions": [ {{ "title": "", "department": "", "urgency": "low|medium|high|urgent", "reason": "" }} ], "summary": "", "classificationReasons": [], "redFlags": [], "growthSignals": [], "benefitsOfProgram": [], "benefitsOfAgents": [] }}, "warnings": [] }} Rules: - Use the SME story and uploaded document text only. - Do not invent registration numbers, tax numbers, IDs, turnover, employees, or compliance status. - If a field is missing, add it to missingFields using a friendly conversational question. - Required basic documents are: Director ID, CIPC document, SARS Tax PIN. - If a required document is not uploaded or cannot be identified, add it to missingDocuments. - Classify the SME stage: - idea: concept only, no operating history - startup: newly operating or still proving model - early_growth: some traction, some customers/revenue - struggling: stagnant revenue, compliance gaps, cashflow pressure, no growth, operational distress - scaling: growing and needing structured support - market_ready: compliant and ready for procurement/market linkage - Urgency must reflect intervention need, not emotional tone. - urgent/high urgency should be used where compliance, tax, funding, payroll, legal, safety, or survival risks are clear. - Recommend interventions only from the intervention catalog where possible. - Recommend one program from availablePrograms where possible. - Match programmes using eligibilityCriteria first, especially province, sector, gender, age, years of trading, B-BBEE level, youth ownership, black ownership, and female ownership. - If the SME does not clearly meet a programme's eligibility criteria, do not force that programme. - If required eligibility data is missing from the SME story, add friendly questions to missingFields. - If no programme fits, leave recommendedProgramId blank and explain why in summary. - benefitsOfProgram must explain why structured incubation helps. - benefitsOfAgents must explain why direct agent support helps without programme onboarding. - Keep text professional and concise. - Return JSON only. Payload: {json.dumps(payload, ensure_ascii=False)} """.strip() def _normalize_sme_intake_result(raw: Dict[str, Any]) -> Dict[str, Any]: decision = raw.get("decision") or {} allowed_stages = ["idea", "startup", "early_growth", "struggling", "scaling", "market_ready"] allowed_urgency = ["low", "medium", "high", "urgent"] allowed_risk = ["low", "medium", "high"] stage = _norm(decision.get("stage")) urgency = _norm(decision.get("urgencyLevel")) risk = _norm(decision.get("riskLevel")) if stage not in allowed_stages: stage = "startup" if urgency not in allowed_urgency: urgency = "medium" if risk not in allowed_risk: risk = "medium" interventions = [] for item in decision.get("recommendedInterventions") or []: item_urgency = _norm(item.get("urgency")) if item_urgency not in allowed_urgency: item_urgency = urgency title = str(item.get("title") or "").strip() if not title: continue interventions.append({ "title": title, "department": str(item.get("department") or "").strip(), "urgency": item_urgency, "reason": str(item.get("reason") or "").strip(), }) missing_fields = [] for item in raw.get("missingFields") or []: field = str(item.get("field") or "").strip() question = str(item.get("question") or "").strip() if not field or not question: continue missing_fields.append({ "field": field, "label": str(item.get("label") or field).strip(), "question": question, "reason": str(item.get("reason") or "").strip(), }) missing_documents = [] for item in raw.get("missingDocuments") or []: doc_type = str(item.get("type") or "").strip() if not doc_type: continue missing_documents.append({ "type": doc_type, "reason": str(item.get("reason") or "").strip(), }) document_findings = [] for item in raw.get("documentFindings") or []: document_findings.append({ "documentType": str(item.get("documentType") or "other").strip(), "filename": str(item.get("filename") or "").strip(), "extractedFields": item.get("extractedFields") or {}, "confidence": _clamp_pct(item.get("confidence")), "warnings": [ str(x).strip() for x in item.get("warnings", []) if str(x).strip() ], }) return { "detectedLanguages": [ str(x).strip() for x in raw.get("detectedLanguages", []) if str(x).strip() ], "confidence": _clamp_pct(raw.get("confidence")), "extractedProfile": raw.get("extractedProfile") or {}, "extractedApplication": raw.get("extractedApplication") or {}, "documentFindings": document_findings, "missingFields": missing_fields[:12], "missingDocuments": missing_documents[:10], "nextQuestion": str(raw.get("nextQuestion") or "").strip(), "decision": { "stage": stage, "urgencyLevel": urgency, "urgencyScore": _clamp_pct(decision.get("urgencyScore")), "riskLevel": risk, "recommendedProgramId": str(decision.get("recommendedProgramId") or "").strip(), "recommendedProgramName": str(decision.get("recommendedProgramName") or "").strip(), "recommendedInterventions": interventions[:10], "summary": str(decision.get("summary") or "").strip(), "classificationReasons": [ str(x).strip() for x in decision.get("classificationReasons", []) if str(x).strip() ][:10], "redFlags": [ str(x).strip() for x in decision.get("redFlags", []) if str(x).strip() ][:10], "growthSignals": [ str(x).strip() for x in decision.get("growthSignals", []) if str(x).strip() ][:10], "benefitsOfProgram": [ str(x).strip() for x in decision.get("benefitsOfProgram", []) if str(x).strip() ][:6], "benefitsOfAgents": [ str(x).strip() for x in decision.get("benefitsOfAgents", []) if str(x).strip() ][:6], }, "warnings": [ str(x).strip() for x in raw.get("warnings", []) if str(x).strip() ], } def build_assignments_page_context(company_code, user_id, program_id, role, user_context): """ Server-side mirror of the assignments page. Must return: - current user's department - eligible SMEs - each SME's required interventions from diagnosticPlans - confirmation status per department - existing assignedInterventions - available consultants for department - recurring / once-off assignment metadata """ # Fetch: # users / operationsStaff / consultants # departments # diagnosticPlans # applications # participants # interventions # assignedInterventions return { "page": "assignments", "companyCode": company_code, "programId": program_id or "all", "department": {}, "rules": { "smmeMustHaveAcceptedApplication": True, "diagnosticPlanMustBeConfirmedByDepartment": True, "diagnosticPlanMustBeConfirmedBySmme": True, "groupedAssignmentRequiresSharedIntervention": True, "recurringCannotSkipOpenPreviousCycle": True, "onceOffCannotRepeatAfterCompletion": True, }, "smes": [], "interventions": [], "assignments": [], "consultants": [] } # -- route --------------------------------------------------------------- @app.route('/chat', methods=['POST']) def chat(): data = request.json role = data.get('role') user_input = data.get('message') company_code = data.get('companyCode') user_id = data.get('userId') if not role or not user_input or not company_code or not user_id: return jsonify({"error": "Missing role, message, companyCode, or userId"}), 400 # 🔎 Resolve current user's email + participantId (participants has no companyCode; resolve via applications) ctx = resolve_user_context(user_id, company_code) token_map = { "{{participantId}}": ctx.get("participantId"), "{{userEmail}}": ctx.get("email"), "{{userId}}": ctx.get("uid"), } # 🧠 Plan planning_result = determine_data_requirements( user_input, company_code, user_id, participant_email=ctx.get("email"), participant_id=ctx.get("participantId"), ) if "error" in planning_result: return jsonify({"reply": f"⚠️ Planning error: {planning_result['error']}"}) # 🛠️ Normalize & replace tokens (schema-aware) normalized_plan = normalize_plan(planning_result, token_map) # 📥 Fetch firestore_data = fetch_data_from_firestore(normalized_plan) # 🧩 Build messages — convert Firestore payload to JSON-safe types system_msg = build_system_message(company_code) safe_ctx = to_jsonable(ctx) safe_data = to_jsonable(firestore_data) data_msg = { "role": "system", "content": ( f"CurrentUserContext: {json.dumps(safe_ctx)}\n" f"Here is the data from Firestore:\n{json.dumps(safe_data)}" ) } user_msg = {"role": "user", "content": user_input} final_response = ask_gpt([system_msg, data_msg, user_msg]) return jsonify({"reply": final_response}) @app.route("/page-assist", methods=["POST"]) def page_assist(): data = request.json or {} page = data.get("page") message = data.get("message") role = data.get("role") company_code = data.get("companyCode") user_id = data.get("userId") program_id = data.get("programId") # optional, "all" allowed if not page or not message or not role or not company_code or not user_id: return jsonify({"error": "Missing page, message, role, companyCode, or userId"}), 400 if page != "assignments": return jsonify({"error": "Unsupported page"}), 400 ctx = resolve_user_context(user_id, company_code) assignment_context = build_assignments_page_context( company_code=company_code, user_id=user_id, program_id=None if program_id in [None, "", "all"] else program_id, role=role, user_context=ctx, ) system_msg = { "role": "system", "content": """ You are a page-aware assistant for the Interventions Assignments page. Answer using ONLY the supplied page context. Explain blockers clearly. Do not invent SMEs, interventions, departments, or assignment rules. If asked why an intervention cannot be assigned, identify the exact reason: - not part of that SME's Diagnostic Plan - Diagnostic Plan not confirmed by both department and SME - not shared across selected SMEs - previous recurring cycle still open - once-off intervention already completed - missing facilitator, due date, target mode, or sub-intervention Return concise JSON: { "reply": "", "reasonCode": "", "affectedSmes": [], "eligibleSmes": [], "blockedSmes": [], "suggestedNextAction": "" } """ } user_msg = { "role": "user", "content": json.dumps({ "question": message, "pageContext": assignment_context, }, ensure_ascii=False) } ai_raw = ask_gpt([system_msg, user_msg]) result = _extract_json_block(ai_raw) return jsonify(to_jsonable(result)) @app.route("/assignments-agent", methods=["POST"]) def assignments_agent(): data = request.json or {} company_code = data.get("companyCode") user_id = data.get("userId") role = data.get("role") message = data.get("message") or data.get("transcript") program_id = data.get("programId") if not company_code or not user_id or not role or not message: return jsonify({"error": "Missing companyCode, userId, role, or message"}), 400 ctx = resolve_user_context(user_id, company_code) assignment_context = build_assignments_page_context( company_code=company_code, user_id=user_id, program_id=None if program_id in [None, "", "all"] else program_id, role=role, user_context=ctx, ) system_msg = { "role": "system", "content": """ You are an assignment action planner. Extract the user's intent, but DO NOT save anything directly. Return strict JSON: { "intent": "assign_intervention|explain|unknown", "interventionName": "", "smeNames": [], "assignmentType": "singular|grouped|unknown", "assignTo": "consultant|operations|unknown", "consultantName": "", "dueDate": "", "targetMode": "hours|sessions|documents|milestones|percentage|unknown", "missingFields": [], "canExecute": false, "validationErrors": [], "confirmationMessage": "" } Required fields before execution: - interventionName - at least one SME - assignTo - dueDate - targetMode - consultantName if assignTo is consultant Validate against pageContext. """ } user_msg = { "role": "user", "content": json.dumps({ "message": message, "pageContext": assignment_context, }, ensure_ascii=False) } ai_raw = ask_gpt([system_msg, user_msg]) plan = _extract_json_block(ai_raw) if not plan.get("canExecute"): return jsonify({ "mode": "needs_more_info", "plan": plan, "reply": plan.get("confirmationMessage") or "I need more information before creating the assignment." }) validation = validate_assignment_plan(plan, assignment_context) if not validation["ok"]: return jsonify({ "mode": "blocked", "reply": validation["message"], "validationErrors": validation["errors"], "plan": plan }) # Do not write immediately unless frontend sends confirmed=true if not data.get("confirmed"): return jsonify({ "mode": "confirmation_required", "reply": validation["confirmationMessage"], "plan": plan }) saved = create_assignments_from_plan( plan=plan, company_code=company_code, user_id=user_id, user_context=ctx, assignment_context=assignment_context, ) return jsonify({ "mode": "executed", "reply": f"Created {saved['count']} assignment(s) successfully.", "createdIds": saved["ids"] }) # --- NEW: lightweight Help/Assistant intent router ------------------------- @app.route('/assist', methods=['POST']) def assist(): """ Classifies a short help question and, if appropriate, returns a suggested navigation target + tour key for the frontend to action. Request JSON: { "role": "incubatee" | ..., "message": "Where do I apply?", "companyCode": "ACME", "userId": "abc123", "path": "/current/location" # optional, helps decide replace vs push if you want } Response JSON: { "reply": "bot text to show", "navigate": { "to": "/incubatee/sme", "tour": "apply", # optional "ts": 1712345678901, # force re-run effects "delayMs": 900 # UI should wait before navigating (to show the bot msg) }, "handoff": { "open": false, # whether to auto-open your drawer after nav "botText": "You're on Programs. Tap an 'Apply' button to begin.", "followups": ["track_application","profile_setup","inquiries"] } } """ data = request.json or {} role = data.get('role') user_input = (data.get('message') or '').strip().lower() company_code = data.get('companyCode') user_id = data.get('userId') if not role or not user_input or not company_code or not user_id: return jsonify({"error": "Missing role, message, companyCode, or userId"}), 400 # Simple keyword intent router (deterministic; fast) def nav_payload(to: str, tour: Optional[str], reply: str, handoff: Optional[Dict[str, Any]] = None): return jsonify({ "reply": reply, "navigate": { "to": to, "tour": tour, "ts": int(datetime.now().timestamp() * 1000), "delayMs": 900 }, "handoff": handoff or {} }) tokens = set(user_input.replace('?', '').split()) contains = lambda *words: any(w in tokens or w in user_input for w in words) # 1) Where do I apply? if contains('apply', 'application', 'register', 'program', 'submit'): return nav_payload( to="/incubatee/sme", tour="apply", reply="Taking you to Programs and starting the tour…", handoff={ "open": False, "botText": "You're on Programs. Tap an **Apply** button to begin.", "followups": ["track_application", "profile_setup", "inquiries"] } ) # 2) Track my application if contains('track', 'status', 'progress'): return nav_payload( to="/incubatee/tracker", tour="track", reply="Opening your application tracker…", handoff={ "open": False, "botText": "This is your tracker. You can view statuses and details here.", "followups": ["where_apply", "inquiries"] } ) # 3) Inquiries / helpdesk if contains('inquiry', 'inquiries', 'question', 'helpdesk', 'support', 'ticket'): return nav_payload( to="/incubatee/sme/inquiries", tour="inquiries", reply="Taking you to Inquiries…", handoff={ "open": True, "botText": "Here you can log questions or check replies.", "followups": ["where_apply", "track_application"] } ) # 4) Profile/setup questions if contains('profile', 'setup', 'account', 'redirect'): return jsonify({ "reply": ( "You need to complete your profile before applying. " "If something’s missing (e.g., name/company), we’ll redirect you to finish it first." ), "handoff": { "open": True, "botText": "Would you like to go to **My Profile** now?", "followups": ["where_apply"] } }) # 5) Contact support if contains('contact', 'email', 'help', 'assist'): return jsonify({ "reply": "You can email support at **support@smartincubation.example** or log an inquiry from the Inquiries page.", "handoff": { "open": True, "botText": "Try the **Inquiries** option if you want to log a ticket.", "followups": ["inquiries"] } }) # 6) Fallback → optionally leverage GPT to draft a helpful answer (no redirect) # Keep this light; we're not doing Firestore fetches here. try: sys = { "role": "system", "content": ( "You are Q-Bot, a concise, friendly assistant. " "Clarify the request in one or two sentences and suggest a next step. " "Do NOT mention databases or internal logic." ) } user = {"role": "user", "content": data.get('message', '')} answer = ask_gpt([sys, user]) or "I’m here to help! Could you tell me a bit more about what you need?" return jsonify({ "reply": answer, "handoff": { "open": True, "followups": ["where_apply", "track_application", "inquiries"] } }) except Exception: return jsonify({ "reply": "I’m here to help! Could you tell me a bit more about what you need?", "handoff": { "open": True, "followups": ["where_apply", "track_application", "inquiries"] } }) @app.route('/analyze-intervention-update', methods=['POST']) def analyze_intervention_update(): """ AI-assisted intervention update analysis. Request JSON: { "role": "consultant", "companyCode": "ABC", "userId": "uid123", "captureMode": "message|notes|audio", "rawInput": "what the user typed or transcript", "intervention": { "id": "...", "interventionTitle": "...", "subtitle": "...", "beneficiaryName": "...", "status": "assigned|in-progress|completed", "targetType": "percentage|number", "targetMetric": "sessions|hours|reports|...", "targetValue": 10, "targetActual": 2, "progress": 20, "dueDate": "...", "notes": "..." }, "history": [ { "notes": "...", "progressAfter": 20, "targetActualAfter": 2, "createdAt": "..." } ] } """ data = request.json or {} role = data.get('role') company_code = data.get('companyCode') user_id = data.get('userId') capture_mode = data.get('captureMode') raw_input = (data.get('rawInput') or '').strip() intervention = data.get('intervention') or {} history = data.get('history') or [] if not role or not company_code or not user_id or not capture_mode or not raw_input: return jsonify({ "error": "Missing role, companyCode, userId, captureMode, or rawInput" }), 400 if capture_mode not in ['message', 'notes', 'audio']: return jsonify({"error": "Invalid captureMode"}), 400 metric_key = _normalize_metric(intervention.get("targetMetric")) metric_rules = _proof_rules_for_metric(metric_key) prompt_payload = { "captureMode": capture_mode, "rawInput": raw_input, "role": role, "intervention": { "id": intervention.get("id"), "interventionTitle": intervention.get("interventionTitle"), "subtitle": intervention.get("subtitle"), "beneficiaryName": intervention.get("beneficiaryName"), "status": intervention.get("status"), "targetType": intervention.get("targetType"), "targetMetric": intervention.get("targetMetric"), "targetValue": intervention.get("targetValue"), "targetActual": intervention.get("targetActual"), "progress": intervention.get("progress"), "dueDate": to_jsonable(intervention.get("dueDate")), "notes": intervention.get("notes"), }, "history": history[:10], "rules": { "metricKey": metric_key, "proofRules": metric_rules, "instruction": ( "Infer what work was likely done from the raw input, " "but do not freely calculate final progress. " "Only suggest hours and measurable units if grounded in the text." ) } } try: system_msg = { "role": "system", "content": ( "You are an assistant that analyses intervention update notes for a business workflow. " "You must return strict JSON only." ) } user_msg = { "role": "user", "content": _build_intervention_analysis_prompt(prompt_payload) } ai_raw = ask_gpt([system_msg, user_msg]) ai_result = _extract_json_block(ai_raw) progress_calc = _calculate_progress_suggestion(intervention, ai_result) merged_proof = _merge_proof_hints(metric_rules, ai_result.get("proofHints", [])) completion_signal = _norm(ai_result.get("completionSignal")) readiness = "not_ready" if completion_signal == "strong": readiness = "close" elif progress_calc.get("suggestedProgressPct", 0) >= 100: readiness = "ready" response = { "summary": ai_result.get("summary", ""), "polishedNotes": ai_result.get("polishedNotes", ""), "activities": ai_result.get("activities", []), "outputs": ai_result.get("outputs", []), "blockers": ai_result.get("blockers", []), "nextSteps": ai_result.get("nextSteps", []), "warnings": ai_result.get("warnings", []), "confidence": ai_result.get("confidence", 0), "completionSignal": completion_signal or "none", "completionReadiness": readiness, "proofSuggestions": merged_proof, "calculation": progress_calc, "rawInput": raw_input, } return jsonify(to_jsonable(response)) except Exception as e: print("analyze_intervention_update_failed:", e) return jsonify({ "error": "Failed to analyse intervention update" }), 500 @app.route('/generate-course-outline', methods=['POST']) def generate_course_outline(): """ multipart/form-data: - file: pdf or docx - role: operations | consultant | admin | ... - companyCode: ... - userId: ... Response: { "reply": "Course outline generated successfully", "outline": { "courseTitle": "...", "courseDescription": "...", "difficulty": "...", "category": "...", "courseType": "...", "estimatedTotalDuration": "...", "learningObjectives": [], "modules": [], "warnings": [] }, "meta": { "filename": "...", "contentType": "...", "extractedChars": 12345, "truncated": false } } """ try: role = request.form.get('role') company_code = request.form.get('companyCode') user_id = request.form.get('userId') uploaded = request.files.get('file') if not role or not company_code or not user_id: return jsonify({ "error": "Missing role, companyCode, or userId" }), 400 if uploaded is None: return jsonify({ "error": "Missing file" }), 400 filename = uploaded.filename or "" if not _allowed_course_source(filename): return jsonify({ "error": "Only PDF and DOCX files are supported" }), 400 file_bytes = uploaded.read() if not file_bytes: return jsonify({ "error": "Uploaded file is empty" }), 400 extracted_text = _extract_course_source_text(filename, file_bytes) if not extracted_text: return jsonify({ "error": "Could not extract readable text from the uploaded file" }), 400 truncated_text = _truncate_source_text(extracted_text) was_truncated = len(truncated_text) < len(extracted_text) system_msg = { "role": "system", "content": ( "You generate practical LMS course outlines from uploaded documents. " "Return strict JSON only." ) } user_msg = { "role": "user", "content": _build_course_outline_prompt(truncated_text, filename) } ai_raw = ask_gpt([system_msg, user_msg]) ai_result = _extract_json_block(ai_raw) outline = _normalize_outline_json(ai_result) if was_truncated: outline["warnings"] = outline.get("warnings", []) outline["warnings"].append( "The source document was long, so only the first portion was used to generate this outline." ) return jsonify({ "reply": "Course outline generated successfully", "outline": outline, "meta": { "filename": filename, "contentType": uploaded.content_type, "extractedChars": len(extracted_text), "truncated": was_truncated, } }) except Exception as e: print("generate_course_outline_failed:", e) return jsonify({ "error": "Failed to generate course outline from file" }), 500 @app.route('/analyze-sme-application-intake', methods=['POST']) def analyze_sme_application_intake(): """ Multipart form-data endpoint for unregistered SME intake. Expected form fields: - mode: initial_review | final_decision - companyCode - userId optional - contactName - contactEmail - contactPhone - rawStory - missingAnswersJson optional - requiredDocumentsJson optional - files[] optional Response: { detectedLanguages, confidence, extractedProfile, extractedApplication, documentFindings, missingFields, missingDocuments, nextQuestion, decision, warnings } """ try: mode = request.form.get("mode") or "initial_review" company_code = (request.form.get("companyCode") or "").strip() user_id = request.form.get("userId") or "" contact_name = request.form.get("contactName") or "" contact_email = request.form.get("email") or request.form.get("contactEmail") or "" contact_phone = request.form.get("contactPhone") or "" raw_story = (request.form.get("rawStory") or "").strip() # companyCode is optional for public/unregistered SME intake if not raw_story: return jsonify({"error": "Missing rawStory"}), 400 try: missing_answers = json.loads(request.form.get("missingAnswersJson") or "{}") except Exception: missing_answers = {} try: required_documents = json.loads(request.form.get("requiredDocumentsJson") or "[]") except Exception: required_documents = [] uploaded_files = request.files.getlist("files") document_payloads = [] for uploaded in uploaded_files: filename = uploaded.filename or "" if not filename: continue if not _allowed_sme_doc(filename): document_payloads.append({ "filename": filename, "contentType": uploaded.content_type, "extractedText": "", "warnings": ["Unsupported document type."] }) continue file_bytes = uploaded.read() if not file_bytes: document_payloads.append({ "filename": filename, "contentType": uploaded.content_type, "extractedText": "", "warnings": ["Uploaded file was empty."] }) continue extracted_text = _extract_sme_doc_text(filename, file_bytes) truncated_text = _truncate_source_text(extracted_text, MAX_SME_DOC_TEXT_CHARS) warnings = [] ext = filename.rsplit(".", 1)[1].lower() if ext in ["png", "jpg", "jpeg"]: warnings.append("Image OCR is not enabled on this endpoint yet.") if extracted_text and len(truncated_text) < len(extracted_text): warnings.append("Document text was truncated before AI analysis.") if not extracted_text and ext not in ["png", "jpg", "jpeg"]: warnings.append("No readable text could be extracted from this document.") document_payloads.append({ "filename": filename, "contentType": uploaded.content_type, "extractedText": truncated_text, "warnings": warnings }) available_programs = _fetch_active_programs(company_code) intervention_catalog = _fetch_intervention_catalog(company_code) payload = { "mode": mode, "companyCode": company_code, "userId": user_id, "contact": { "name": contact_name, "email": contact_email, "phone": contact_phone, }, "rawStory": raw_story, "missingAnswers": missing_answers, "requiredDocuments": required_documents, "uploadedDocuments": document_payloads, "availablePrograms": available_programs, "interventionCatalog": intervention_catalog, "analysisDate": datetime.utcnow().isoformat(), } system_msg = { "role": "system", "content": ( "You analyse SME intake information for an incubation platform. " "You return strict JSON only. " "You must not invent official facts, registration numbers, tax numbers, " "document statuses, revenue, employees, or compliance claims." ) } user_msg = { "role": "user", "content": _build_sme_intake_prompt(payload) } ai_raw = ask_gpt([system_msg, user_msg]) ai_result = _extract_json_block(ai_raw) normalized = _normalize_sme_intake_result(ai_result) return jsonify(to_jsonable(normalized)) except Exception as e: print("analyze_sme_application_intake_failed:", e) return jsonify({ "error": "Failed to analyse SME application intake" }), 500 # Assessment extraction (uses this deployment's Firebase project and Gemini client) from flask_assessment_extraction import create_assessment_blueprint app.register_blueprint(create_assessment_blueprint(ask_gpt, db)) if __name__ == "__main__": app.run(host="0.0.0.0", port=7860)