418 lines
16 KiB
Python
418 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
api.py — Forensic Features API
|
|
|
|
Singurul scop al acestui serviciu: extrage măsurători forensice obiective
|
|
din video/imagini și le returnează într-un format consumabil de un LLM
|
|
extern (Qwen Vision, GPT-4V, Claude, etc.) care face deja vision/OCR pe
|
|
imaginile originale.
|
|
|
|
Endpoint-uri:
|
|
POST /api/forensic-evidence — upload video/imagine, returnează
|
|
evidence_text + base64 PNG-uri + raw scores
|
|
GET /api/forensic-modules — listează modulele disponibile (m25-m29)
|
|
GET /api/status/{job_id} — polling pentru cereri async
|
|
GET /api/result/{job_id} — preluare rezultat job async
|
|
GET /health — healthcheck
|
|
|
|
Vezi docs/API.md pentru detalii complete.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import aiohttp
|
|
from aiohttp import web
|
|
import aiohttp_cors
|
|
|
|
BASE_DIR = Path(__file__).parent
|
|
sys.path.insert(0, str(BASE_DIR))
|
|
|
|
INFER_DIR = BASE_DIR / "data" / "inference"
|
|
RESULTS_DIR = BASE_DIR / "data" / "results"
|
|
|
|
from forensic.orchestrator import (
|
|
run_forensic_pipeline, AVAILABLE_MODULES,
|
|
)
|
|
from forensic.prompt_builder import build_evidence_block
|
|
|
|
# ── Job store in-memory pentru cereri async ──────────────────────────────
|
|
_jobs: dict[str, dict] = {}
|
|
|
|
|
|
# ───────────────────────────────────────────────────────────────────────────
|
|
# Core pipeline runner — apelat sync sau via executor în handler async
|
|
# ───────────────────────────────────────────────────────────────────────────
|
|
|
|
def _run_forensic_and_format(
|
|
video_path: str, results_dir: str,
|
|
modules: list[str] | None, every_n_frames: int | None,
|
|
encode_images: bool,
|
|
) -> dict:
|
|
"""
|
|
Rulează orchestratorul + prompt builder. Sync — apelat în executor thread.
|
|
|
|
every_n_frames=None → orchestrator calculează adaptive bazat pe durata video.
|
|
"""
|
|
orch = run_forensic_pipeline(
|
|
video_path=video_path,
|
|
results_dir=results_dir,
|
|
modules=modules,
|
|
every_n_frames=every_n_frames,
|
|
use_parallel=False,
|
|
)
|
|
|
|
images_dir = os.path.join(results_dir, "images")
|
|
evidence = build_evidence_block(
|
|
module_results=orch["modules"],
|
|
fusion=orch["fusion"],
|
|
images_dir=images_dir,
|
|
encode_images=encode_images,
|
|
include_instruction=True,
|
|
)
|
|
|
|
return {
|
|
"video_path": os.path.basename(video_path),
|
|
"n_frames_extracted": orch["n_frames_extracted"],
|
|
"every_n_frames_used": orch.get("every_n_frames_used"),
|
|
"auto_skipped": orch.get("auto_skipped", []),
|
|
"modules_run": list(orch["modules"].keys()),
|
|
"execution_time_ms": orch["execution_time_ms"],
|
|
"fusion": orch["fusion"],
|
|
"explanation": orch["explanation"],
|
|
"modules": orch["modules"],
|
|
"evidence_text": evidence["evidence_text"],
|
|
"images": evidence["images"],
|
|
"summary": evidence["summary"],
|
|
"instruction_for_llm": evidence["instruction_for_llm"],
|
|
"errors": orch["errors"],
|
|
}
|
|
|
|
|
|
def _async_pipeline_wrapper(
|
|
job_id: str, video_path: str, results_dir: str,
|
|
modules: list[str], every_n_frames: int | None, encode_images: bool,
|
|
) -> None:
|
|
"""Wrapper executor pentru cereri async — update-uri în _jobs."""
|
|
try:
|
|
_jobs[job_id].update(
|
|
status="running",
|
|
progress="Running forensic modules m25-m29...",
|
|
)
|
|
result = _run_forensic_and_format(
|
|
video_path, results_dir, modules, every_n_frames, encode_images
|
|
)
|
|
_jobs[job_id].update(status="done", result=result, progress="Complete.")
|
|
except Exception as e:
|
|
_jobs[job_id].update(status="error", error=str(e), progress=f"Error: {e}")
|
|
finally:
|
|
try:
|
|
up = Path(video_path)
|
|
if up.exists():
|
|
up.unlink()
|
|
if up.parent.exists() and not any(up.parent.iterdir()):
|
|
up.parent.rmdir()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ───────────────────────────────────────────────────────────────────────────
|
|
# HTTP handlers
|
|
# ───────────────────────────────────────────────────────────────────────────
|
|
|
|
async def handle_forensic_evidence(request: web.Request) -> web.Response:
|
|
"""
|
|
POST /api/forensic-evidence — multipart/form-data.
|
|
|
|
Form fields:
|
|
video (file, required) Video sau imagine de analizat
|
|
(mp4, mov, avi, mkv, webm, jpg, png).
|
|
modules (str, optional) CSV de module IDs ("m25,m27,m28").
|
|
Default: toate cele 5 module.
|
|
encode_images (str, optional) "1" (default) = atașează data URLs
|
|
base64 pentru PNG-uri.
|
|
"0" = doar paths.
|
|
every_n_frames (int, optional) Pas extracție cadre. Default: adaptiv
|
|
după durata video.
|
|
async_mode (str, optional) "0" (default) = sync, returnează rezultat.
|
|
"1" = creează job + returnează 202.
|
|
|
|
Returns:
|
|
Sync (200) — JSON cu:
|
|
evidence_text, images, summary, modules, fusion,
|
|
instruction_for_llm — vezi docs/CONTRACT.md.
|
|
Async (202) — {"job_id": "...", "status": "queued"}.
|
|
"""
|
|
reader = await request.multipart()
|
|
|
|
video_field = None
|
|
modules_csv = None
|
|
encode_images = True
|
|
every_n_frames: int | None = None
|
|
async_mode = False
|
|
|
|
while True:
|
|
field = await reader.next()
|
|
if field is None:
|
|
break
|
|
if field.name == "video":
|
|
video_field = field
|
|
break
|
|
elif field.name == "modules":
|
|
modules_csv = (await field.text()).strip()
|
|
elif field.name == "encode_images":
|
|
encode_images = (await field.text()).strip() not in ("0", "false", "no", "")
|
|
elif field.name == "every_n_frames":
|
|
try:
|
|
every_n_frames = int((await field.text()).strip())
|
|
except ValueError:
|
|
pass
|
|
elif field.name == "async_mode":
|
|
async_mode = (await field.text()).strip() in ("1", "true", "yes")
|
|
|
|
if video_field is None:
|
|
raise web.HTTPBadRequest(text="Field 'video' is required in multipart upload.")
|
|
|
|
filename = video_field.filename or ""
|
|
allowed = (".mp4", ".mov", ".avi", ".mkv", ".webm", ".jpg", ".jpeg", ".png")
|
|
if not filename.lower().endswith(allowed):
|
|
raise web.HTTPBadRequest(
|
|
text=f"Unsupported file type: '{filename}'. Allowed: {', '.join(allowed)}."
|
|
)
|
|
|
|
# Validate module list
|
|
if modules_csv:
|
|
modules_requested = [m.strip() for m in modules_csv.split(",") if m.strip()]
|
|
invalid = [m for m in modules_requested if m not in AVAILABLE_MODULES]
|
|
if invalid:
|
|
raise web.HTTPBadRequest(
|
|
text=f"Unknown modules: {invalid}. Available: {list(AVAILABLE_MODULES.keys())}"
|
|
)
|
|
else:
|
|
modules_requested = list(AVAILABLE_MODULES.keys())
|
|
|
|
# Save upload to disk
|
|
job_id = uuid.uuid4().hex[:16]
|
|
work_dir = INFER_DIR / job_id
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
upload_path = work_dir / f"{job_id}_input{Path(filename).suffix.lower()}"
|
|
|
|
with open(upload_path, "wb") as f:
|
|
while True:
|
|
chunk = await video_field.read_chunk(65536)
|
|
if not chunk:
|
|
break
|
|
f.write(chunk)
|
|
|
|
results_dir = RESULTS_DIR / job_id
|
|
results_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# ── Async mode: queue job and return immediately ──
|
|
if async_mode:
|
|
_jobs[job_id] = {
|
|
"status": "queued",
|
|
"progress": "Forensic evidence pipeline queued",
|
|
"result": None,
|
|
"error": None,
|
|
}
|
|
loop = asyncio.get_event_loop()
|
|
loop.run_in_executor(
|
|
None, _async_pipeline_wrapper,
|
|
job_id, str(upload_path), str(results_dir),
|
|
modules_requested, every_n_frames, encode_images,
|
|
)
|
|
return web.json_response(
|
|
{"job_id": job_id, "status": "queued", "modules": modules_requested},
|
|
status=202,
|
|
)
|
|
|
|
# ── Sync mode: run inline and return result ──
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
result = await loop.run_in_executor(
|
|
None, _run_forensic_and_format,
|
|
str(upload_path), str(results_dir),
|
|
modules_requested, every_n_frames, encode_images,
|
|
)
|
|
return web.json_response(result)
|
|
except Exception as e:
|
|
return web.json_response(
|
|
{"error": str(e), "modules_requested": modules_requested},
|
|
status=500,
|
|
)
|
|
finally:
|
|
try:
|
|
if upload_path.exists():
|
|
upload_path.unlink()
|
|
if work_dir.exists() and not any(work_dir.iterdir()):
|
|
work_dir.rmdir()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
async def handle_forensic_modules(request: web.Request) -> web.Response:
|
|
"""GET /api/forensic-modules — listează modulele disponibile."""
|
|
catalog = []
|
|
for mid, (import_path, fn_name, input_type) in AVAILABLE_MODULES.items():
|
|
catalog.append({
|
|
"id": mid,
|
|
"input_type": input_type,
|
|
"import_path": import_path,
|
|
"function": fn_name,
|
|
})
|
|
return web.json_response({
|
|
"available_modules": catalog,
|
|
"default": list(AVAILABLE_MODULES.keys()),
|
|
})
|
|
|
|
|
|
async def handle_status(request: web.Request) -> web.Response:
|
|
"""GET /api/status/{job_id} — polling pentru cereri async."""
|
|
job_id = request.match_info["job_id"]
|
|
if job_id not in _jobs:
|
|
raise web.HTTPNotFound(text=f"Job '{job_id}' not found.")
|
|
job = _jobs[job_id]
|
|
return web.json_response({
|
|
"job_id": job_id,
|
|
"status": job["status"],
|
|
"progress": job["progress"],
|
|
})
|
|
|
|
|
|
async def handle_result(request: web.Request) -> web.Response:
|
|
"""GET /api/result/{job_id} — preluare rezultat job async."""
|
|
job_id = request.match_info["job_id"]
|
|
if job_id not in _jobs:
|
|
raise web.HTTPNotFound(text=f"Job '{job_id}' not found.")
|
|
job = _jobs[job_id]
|
|
if job["status"] == "error":
|
|
return web.json_response({"job_id": job_id, "error": job["error"]}, status=500)
|
|
if job["status"] != "done":
|
|
return web.json_response(
|
|
{"job_id": job_id, "status": job["status"], "progress": job["progress"]},
|
|
status=202,
|
|
)
|
|
return web.json_response(job["result"])
|
|
|
|
|
|
async def handle_health(request: web.Request) -> web.Response:
|
|
"""GET /health — healthcheck pentru Docker / load balancer."""
|
|
return web.json_response({
|
|
"status": "ok",
|
|
"service": "forensic-features",
|
|
"modules": list(AVAILABLE_MODULES.keys()),
|
|
})
|
|
|
|
|
|
# ───────────────────────────────────────────────────────────────────────────
|
|
# App setup
|
|
# ───────────────────────────────────────────────────────────────────────────
|
|
|
|
async def handle_metrics(_request: web.Request) -> web.Response:
|
|
"""Prometheus /metrics endpoint."""
|
|
try:
|
|
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST # type: ignore
|
|
return web.Response(body=generate_latest(), content_type=CONTENT_TYPE_LATEST.split(';')[0])
|
|
except ImportError:
|
|
return web.Response(text="prometheus_client not installed\n", status=503)
|
|
|
|
|
|
def build_app() -> web.Application:
|
|
app = web.Application(client_max_size=2 * 1024**3) # 2GB upload limit
|
|
|
|
app.router.add_post("/api/forensic-evidence", handle_forensic_evidence)
|
|
app.router.add_get( "/api/forensic-modules", handle_forensic_modules)
|
|
app.router.add_get( "/api/status/{job_id}", handle_status)
|
|
app.router.add_get( "/api/result/{job_id}", handle_result)
|
|
app.router.add_get( "/health", handle_health)
|
|
app.router.add_get( "/metrics", handle_metrics)
|
|
|
|
cors = aiohttp_cors.setup(app, defaults={
|
|
"*": aiohttp_cors.ResourceOptions(
|
|
allow_credentials=True,
|
|
expose_headers="*",
|
|
allow_headers="*",
|
|
)
|
|
})
|
|
for route in list(app.router.routes()):
|
|
cors.add(route)
|
|
|
|
# HTTP middleware: count requests + duration per route
|
|
try:
|
|
from prometheus_client import Counter, Histogram # type: ignore
|
|
|
|
REQ_COUNTER = Counter("http_requests_total", "Total HTTP requests",
|
|
labelnames=["method", "path", "status"])
|
|
REQ_LATENCY = Histogram("http_request_duration_seconds", "HTTP request duration",
|
|
labelnames=["method", "path"],
|
|
buckets=[0.005, 0.025, 0.1, 0.5, 1, 5, 30, 90, 300])
|
|
import time as _time
|
|
|
|
@web.middleware
|
|
async def metrics_middleware(request: web.Request, handler):
|
|
start = _time.perf_counter()
|
|
try:
|
|
response = await handler(request)
|
|
status = response.status
|
|
return response
|
|
except web.HTTPException as exc:
|
|
status = exc.status
|
|
raise
|
|
except Exception:
|
|
status = 500
|
|
raise
|
|
finally:
|
|
# Use route.resource canonical path so /api/status/{job_id} groups together
|
|
path = request.match_info.route.resource.canonical if request.match_info.route.resource else request.path
|
|
REQ_COUNTER.labels(method=request.method, path=path, status=str(status)).inc()
|
|
REQ_LATENCY.labels(method=request.method, path=path).observe(_time.perf_counter() - start)
|
|
|
|
app.middlewares.append(metrics_middleware)
|
|
except ImportError:
|
|
pass
|
|
|
|
# OTel tracing — aiohttp server + client auto-instrumentation
|
|
import os as _os
|
|
_otel_ep = _os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT")
|
|
if _otel_ep:
|
|
try:
|
|
from opentelemetry import trace as _trace # type: ignore
|
|
from opentelemetry.sdk.resources import Resource as _R # type: ignore
|
|
from opentelemetry.sdk.trace import TracerProvider as _TP # type: ignore
|
|
from opentelemetry.sdk.trace.export import BatchSpanProcessor as _BSP # type: ignore
|
|
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter as _Exp # type: ignore
|
|
from opentelemetry.instrumentation.aiohttp_server import AioHttpServerInstrumentor as _AInst # type: ignore
|
|
|
|
_provider = _TP(resource=_R.create({"service.name": _os.environ.get("OTEL_SERVICE_NAME", "forensic-features-api")}))
|
|
_provider.add_span_processor(_BSP(_Exp(endpoint=_otel_ep, insecure=True)))
|
|
_trace.set_tracer_provider(_provider)
|
|
_AInst().instrument()
|
|
print(f"[otel] forensic-features-api instrumented -> {_otel_ep}")
|
|
except ImportError as _e:
|
|
print(f"[otel] skip: {_e}")
|
|
|
|
return app
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Forensic Features API")
|
|
parser.add_argument("--host", default=os.environ.get("API_HOST", "0.0.0.0"))
|
|
parser.add_argument("--port", type=int,
|
|
default=int(os.environ.get("API_PORT", "8080")))
|
|
args = parser.parse_args()
|
|
|
|
os.chdir(BASE_DIR)
|
|
INFER_DIR.mkdir(parents=True, exist_ok=True)
|
|
RESULTS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
print(f"Forensic Features API on http://{args.host}:{args.port}")
|
|
print(f"Available modules: {list(AVAILABLE_MODULES.keys())}")
|
|
web.run_app(build_app(), host=args.host, port=args.port)
|