Spaces:
Running
Running
| """ | |
| BharatGraph - Phase 34: Forensic Intelligence API | |
| GET /forensics/circular-ownership -- detect shell company ownership rings | |
| GET /forensics/ghost-companies -- companies with contracts but no activity | |
| GET /forensics/shadow-directors -- directors of 10+ companies / shared addresses | |
| GET /forensics/benfords/{entity_id} -- Benford Law analysis on asset declarations | |
| GET /forensics/shadow-draft -- policy text similarity to lobbying documents | |
| Pure ASCII. All detectors built in ai/ -- this file just exposes them. | |
| """ | |
| import os, sys | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) | |
| from datetime import datetime | |
| from fastapi import APIRouter, Depends, Query | |
| from loguru import logger | |
| from api.dependencies import get_db | |
| router = APIRouter(prefix="/forensics", tags=["Forensics"]) | |
| def circular_ownership( | |
| max_cycle_length: int = Query(6, ge=3, le=10), | |
| driver=Depends(get_db), | |
| ): | |
| """ | |
| Detect shell company ownership cycles in the full graph. | |
| A cycle A -> B -> C -> A indicates potential circular ownership | |
| used to obscure beneficial ownership. | |
| """ | |
| logger.info("[Forensics] circular ownership scan") | |
| try: | |
| from ai.circular_ownership import CircularOwnershipDetector | |
| det = CircularOwnershipDetector(driver=driver) | |
| cycles = det.detect_cycles() | |
| filtered = [c for c in cycles if len(c.get("cycle", [])) <= max_cycle_length] | |
| return { | |
| "total_cycles_detected": len(cycles), | |
| "shown": len(filtered), | |
| "max_cycle_length": max_cycle_length, | |
| "cycles": filtered, | |
| "analyzed_at": datetime.now().isoformat(), | |
| "note": ( | |
| "Circular ownership often indicates shell company structures " | |
| " used to obscure beneficial ownership or inflate valuations." | |
| ), | |
| } | |
| except Exception as e: | |
| logger.error(f"[Forensics] circular ownership error: {type(e).__name__}") | |
| return {"status": "error", "detail": str(type(e).__name__), | |
| "analyzed_at": datetime.now().isoformat()} | |
| def ghost_companies( | |
| min_score: int = Query(70, ge=0, le=100), | |
| limit: int = Query(50, ge=1, le=200), | |
| driver=Depends(get_db), | |
| ): | |
| """ | |
| Score every company in the graph for ghost company indicators: | |
| - No declared employees | |
| - Minimal registered capital | |
| - High government contract volume | |
| - No web presence or filings | |
| Returns companies with ghost_score >= min_score (default 70). | |
| """ | |
| logger.info(f"[Forensics] ghost company scan min_score={min_score}") | |
| try: | |
| from ai.ghost_company import GhostCompanyDetector | |
| det = GhostCompanyDetector(driver=driver) | |
| results = det.run_detection() | |
| flagged = [r for r in results if r.get("ghost_score", 0) >= min_score] | |
| flagged.sort(key=lambda x: x.get("ghost_score", 0), reverse=True) | |
| return { | |
| "total_scanned": len(results), | |
| "flagged_count": len(flagged), | |
| "min_score": min_score, | |
| "companies": flagged[:limit], | |
| "analyzed_at": datetime.now().isoformat(), | |
| } | |
| except Exception as e: | |
| logger.error(f"[Forensics] ghost companies error: {type(e).__name__}") | |
| return {"status": "error", "detail": str(type(e).__name__), | |
| "analyzed_at": datetime.now().isoformat()} | |
| def shadow_directors( | |
| min_company_count: int = Query(10, ge=3, le=100), | |
| driver=Depends(get_db), | |
| ): | |
| """ | |
| Detect shadow director patterns: | |
| 1. Individuals who are director of 10+ companies | |
| 2. Multiple companies sharing the same registered address | |
| (indicates a registration agent acting as nominee director) | |
| """ | |
| logger.info("[Forensics] shadow director scan") | |
| try: | |
| from ai.shadow_director import ShadowDirectorDetector | |
| det = ShadowDirectorDetector(driver=driver) | |
| result = det.run_full_detection() | |
| high_count = [ | |
| r for r in result.get("high_directorship_count", []) | |
| if r.get("company_count", 0) >= min_company_count | |
| ] | |
| return { | |
| "address_reuse_clusters": result.get("address_reuse", []), | |
| "high_directorship_entities": high_count, | |
| "min_company_count": min_company_count, | |
| "analyzed_at": datetime.now().isoformat(), | |
| } | |
| except Exception as e: | |
| logger.error(f"[Forensics] shadow directors error: {type(e).__name__}") | |
| return {"status": "error", "detail": str(type(e).__name__), | |
| "analyzed_at": datetime.now().isoformat()} | |
| def benfords_analysis(entity_id: str, driver=Depends(get_db)): | |
| """ | |
| Run Benford Law analysis on all affidavit asset values declared | |
| by this entity. Significant deviation (chi2 > 15.5) suggests | |
| fabricated or heavily rounded financial figures. | |
| """ | |
| logger.info(f"[Forensics] Benford analysis entity={entity_id[:8]}") | |
| try: | |
| from ai.benfords_analyzer import BenfordsAnalyzer | |
| ba = BenfordsAnalyzer() | |
| with driver.session() as s: | |
| rows = s.run( | |
| "MATCH (n {id:})-[:FILED_AFFIDAVIT]->(a:Affidavit)" | |
| " RETURN a.total_assets_crore AS total," | |
| " a.movable_assets_crore AS movable," | |
| " a.liabilities_crore AS liabilities," | |
| " a.year AS year", | |
| id=entity_id | |
| ).data() | |
| if not rows: | |
| return { | |
| "entity_id": entity_id, | |
| "status": "no_data", | |
| "note": "No affidavit records found for this entity", | |
| } | |
| values = [] | |
| for r in rows: | |
| for k in ("total", "movable", "liabilities"): | |
| if r.get(k): values.append(float(r[k])) | |
| if len(values) < 5: | |
| return { | |
| "entity_id": entity_id, | |
| "status": "insufficient_data", | |
| "value_count": len(values), | |
| "note": "Fewer than 5 numeric values -- Benford analysis unreliable", | |
| } | |
| result = ba.analyze(values) | |
| result["entity_id"] = entity_id | |
| result["value_count"] = len(values) | |
| result["analyzed_at"] = datetime.now().isoformat() | |
| return result | |
| except Exception as e: | |
| logger.error(f"[Forensics] Benford error entity={entity_id[:8]}: {type(e).__name__}") | |
| return {"status": "error", "detail": str(type(e).__name__)} | |
| def shadow_draft_check( | |
| submission_id: str = Query(..., description="Node ID of submitted policy/bill text"), | |
| bill_id: str = Query(..., description="Node ID of reference bill text"), | |
| driver=Depends(get_db), | |
| ): | |
| """ | |
| Compare two policy/bill texts for shadow drafting (text copied from | |
| lobbying documents or industry submissions into government bills). | |
| Returns similarity score and matched sections. | |
| """ | |
| logger.info(f"[Forensics] shadow draft check {submission_id[:8]} vs {bill_id[:8]}") | |
| try: | |
| from ai.shadow_draft_detector import ShadowDraftDetector | |
| det = ShadowDraftDetector() | |
| with driver.session() as s: | |
| sub = s.run( | |
| "MATCH (n {id:}) RETURN coalesce(n.text,n.content,n.summary) AS t", | |
| id=submission_id | |
| ).single() | |
| bill = s.run( | |
| "MATCH (n {id:}) RETURN coalesce(n.text,n.content,n.summary) AS t", | |
| id=bill_id | |
| ).single() | |
| if not sub or not bill: | |
| return {"status": "node_not_found", | |
| "found_submission": sub is not None, | |
| "found_bill": bill is not None} | |
| result = det.compare(sub["t"] or "", bill["t"] or "") | |
| result["submission_id"] = submission_id | |
| result["bill_id"] = bill_id | |
| result["analyzed_at"] = datetime.now().isoformat() | |
| return result | |
| except Exception as e: | |
| logger.error(f"[Forensics] shadow draft error: {type(e).__name__}") | |
| return {"status": "error", "detail": str(type(e).__name__)} | |