"""MedArchive API — эндпоинты ТЗ §4.5 плюс /stats для дашборда. Читает базу, собранную конвейером (etl/pipeline.py). Для MVP это SQLite; путь задаётся переменной окружения MEDARCHIVE_DB. Ядро WOW — /services/{id}/partners: «кто оказывает услугу и по какой цене», отсортировано от выгодной. """ from __future__ import annotations import json import os import sqlite3 import sys from pathlib import Path from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from contracts.models import PartnerOut, PriceOut, ServiceOut, Stats # noqa: E402 DB_PATH = os.environ.get("MEDARCHIVE_DB", str(Path(__file__).resolve().parents[1] / "data/medarchive.db")) app = FastAPI( title="MedArchive API", version="1.0.0", description="Поиск услуг и цен по архиву прайсов клиник-партнёров.", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"] ) def db() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def _price_out(row: sqlite3.Row) -> PriceOut: return PriceOut( partner_id=row["partner_id"], partner_name=row["partner_name"], price_resident_kzt=row["price_resident"], price_nonresident_kzt=row["price_nonresident"], prices=json.loads(row["prices"] or "{}"), effective_date=row["effective_date"], ) @app.get("/services", response_model=list[ServiceOut], summary="Услуги справочника") def list_services(specialty: str | None = None, q: str | None = None, limit: int = 100): sql = "SELECT service_id, specialty, name_ru, tarificator_code FROM service WHERE 1=1" args: list = [] if specialty: sql += " AND specialty = ?" args.append(specialty) if q: sql += " AND name_ru LIKE ?" args.append(f"%{q}%") sql += " ORDER BY name_ru LIMIT ?" args.append(limit) with db() as conn: return [ServiceOut(**dict(r)) for r in conn.execute(sql, args)] @app.get("/services/{service_id}/partners", response_model=list[PriceOut], summary="Кто оказывает услугу и по какой цене") def service_partners(service_id: str): """Одна строка на клинику с показательной ценой (резидент или минимальный тариф).""" sql = """SELECT pi.partner_id, p.name AS partner_name, pi.price_resident, pi.price_nonresident, pi.prices, pi.effective_date FROM price_item pi JOIN partner p ON pi.partner_id = p.partner_id WHERE pi.service_id = ? AND pi.is_active = 1""" best: dict[str, PriceOut] = {} with db() as conn: for r in conn.execute(sql, [service_id]): prices = json.loads(r["prices"] or "{}") display = r["price_resident"] or (min(prices.values()) if prices else None) current = best.get(r["partner_id"]) if current is None or ( display is not None and (current.price_resident_kzt is None or display < current.price_resident_kzt) ): best[r["partner_id"]] = PriceOut( partner_id=r["partner_id"], partner_name=r["partner_name"], price_resident_kzt=display, price_nonresident_kzt=r["price_nonresident"], prices=prices, effective_date=r["effective_date"], ) return sorted(best.values(), key=lambda x: x.price_resident_kzt or float("inf")) @app.get("/partners", response_model=list[PartnerOut], summary="Партнёры") def list_partners(city: str | None = None): sql = "SELECT partner_id, name, city FROM partner WHERE 1=1" args: list = [] if city: sql += " AND city = ?" args.append(city) sql += " ORDER BY name" with db() as conn: return [PartnerOut(partner_id=r["partner_id"], name=r["name"], city=r["city"]) for r in conn.execute(sql, args)] @app.get("/partners/{partner_id}/services", response_model=list[PriceOut], summary="Все услуги партнёра с ценами") def partner_services(partner_id: str, limit: int = 500): sql = """SELECT pi.partner_id, p.name AS partner_name, pi.price_resident, pi.price_nonresident, pi.prices, pi.effective_date FROM price_item pi JOIN partner p ON pi.partner_id = p.partner_id WHERE pi.partner_id = ? AND pi.is_active = 1 LIMIT ?""" with db() as conn: return [_price_out(r) for r in conn.execute(sql, [partner_id, limit])] @app.get("/search", summary="Полнотекстовый поиск по услугам и партнёрам") def search(q: str = Query(..., min_length=1), limit: int = 20): with db() as conn: services = [ dict(r) for r in conn.execute( "SELECT service_id, specialty, name_ru FROM service WHERE name_ru LIKE ? LIMIT ?", [f"%{q}%", limit], ) ] partners = [ dict(r) for r in conn.execute( "SELECT partner_id, name, city FROM partner WHERE name LIKE ? LIMIT ?", [f"%{q}%", limit] ) ] return {"services": services, "partners": partners} @app.get("/unmatched", summary="Несопоставленные позиции (для операторов)") def unmatched(limit: int = 50, offset: int = 0): sql = """SELECT pi.item_id, pi.service_name_raw, pi.prices, p.name AS partner_name FROM price_item pi JOIN partner p ON pi.partner_id = p.partner_id WHERE pi.service_id IS NULL AND pi.is_active = 1 ORDER BY pi.item_id LIMIT ? OFFSET ?""" with db() as conn: return [ { "item_id": r["item_id"], "service_name_raw": r["service_name_raw"], "partner_name": r["partner_name"], "prices": json.loads(r["prices"] or "{}"), } for r in conn.execute(sql, [limit, offset]) ] @app.post("/match", summary="Ручное сопоставление позиции со справочником") def manual_match(item_id: int, service_id: str): with db() as conn: if not conn.execute("SELECT 1 FROM service WHERE service_id = ?", [service_id]).fetchone(): raise HTTPException(404, "услуга справочника не найдена") cur = conn.execute( "UPDATE price_item SET service_id = ?, map_method = 'manual', map_confidence = 1.0 WHERE item_id = ?", [service_id, item_id], ) conn.commit() if cur.rowcount == 0: raise HTTPException(404, "позиция не найдена") return {"item_id": item_id, "service_id": service_id, "status": "matched"} @app.get("/stats", response_model=Stats, summary="Сводка качества обработки") def stats(): report = Path(DB_PATH).parent / "quality_report.json" if report.exists(): data = json.loads(report.read_text(encoding="utf-8")) return Stats( documents_total=data.get("documents", 0), documents_done=data.get("documents", 0), items_total=data.get("items", 0), auto_matched_pct=data.get("auto_matched_pct", 0.0), unmatched_total=data.get("unmatched", 0), ) return Stats() @app.get("/health", summary="Проверка живости") def health(): return {"status": "ok", "db": Path(DB_PATH).exists()}