spb/pdb-api

1"""pdb_client — client Python minimal pour la PDB API (SEC Project Intelligence Database, UQO).23Aucune dépendance obligatoire (urllib). pandas est optionnel pour `to_dataframe`.45    from pdb_client import PDB6    pdb = PDB("VOTRE_CLE")                      # ou variable d'environnement PDB_API_KEY7    pdb.stats()["overview"]8    for p in pdb.iter_projects(type="data_center", min_amount=5e8):9        print(p["ticker"], p["project_name"], p["total_amount_usd"])10    df = pdb.to_dataframe(pdb.iter_projects(sector="Utilities"))11    pdb.sql("select project_type, count(*) n from projects group by 1 order by n desc")12"""13from __future__ import annotations1415import csv16import io17import json18import os19import time20import urllib.error21import urllib.parse22import urllib.request23from typing import Any, Iterator2425__version__ = "1.0.0"26DEFAULT_BASE = "https://www.pdb-api.co/v1"272829class PDBError(RuntimeError):30    def __init__(self, status: int, detail: str, url: str):31        super().__init__(f"HTTP {status} — {detail} ({url})")32        self.status, self.detail, self.url = status, detail, url333435class PDB:36    def __init__(self, api_key: str | None = None, base_url: str = DEFAULT_BASE, timeout: float = 60.0, retries: int = 3):37        self.api_key = api_key or os.environ.get("PDB_API_KEY", "")38        if not self.api_key:39            raise ValueError("Clé d'API manquante : PDB(api_key=...) ou variable PDB_API_KEY.")40        self.base_url = base_url.rstrip("/")41        self.timeout = timeout42        self.retries = retries4344    # ------------------------------------------------------------ transport45    def request(self, method: str, path: str, params: dict | None = None, body: dict | None = None, raw: bool = False) -> Any:46        q = {k: (str(v).lower() if isinstance(v, bool) else v) for k, v in (params or {}).items() if v is not None and v != ""}47        url = f"{self.base_url}/{path.lstrip('/')}" + (f"?{urllib.parse.urlencode(q)}" if q else "")48        data = json.dumps(body).encode() if body is not None else None49        headers = {"X-API-Key": self.api_key, "Accept": "application/json", "User-Agent": f"pdb_client/{__version__}"}50        if data is not None:51            headers["Content-Type"] = "application/json"52        for attempt in range(self.retries + 1):53            req = urllib.request.Request(url, data=data, method=method, headers=headers)54            try:55                with urllib.request.urlopen(req, timeout=self.timeout) as r:56                    payload = r.read()57                    return payload.decode() if raw else json.loads(payload)58            except urllib.error.HTTPError as e:59                detail = e.read().decode(errors="replace")60                try:61                    detail = json.loads(detail).get("detail", detail)62                except Exception:63                    pass64                if e.code in (429, 502, 503, 504) and attempt < self.retries:65                    time.sleep(1.5 * (attempt + 1))66                    continue67                raise PDBError(e.code, detail, url) from None6869    def get(self, path: str, **params) -> Any:70        return self.request("GET", path, params)7172    # ------------------------------------------------------------ découverte73    def health(self) -> dict:74        return self.get("health")7576    def stats(self) -> dict:77        return self.get("stats")7879    def taxonomy(self) -> list[dict]:80        return self.get("taxonomy")8182    def sectors(self) -> list[dict]:83        return self.get("sectors")8485    def technologies(self, q: str | None = None, limit: int = 50) -> list[dict]:86        return self.get("technologies", q=q, limit=limit)8788    def locations(self, q: str | None = None, limit: int = 50) -> list[dict]:89        return self.get("locations", q=q, limit=limit)9091    def partners(self, q: str | None = None, limit: int = 50) -> list[dict]:92        return self.get("partners", q=q, limit=limit)9394    def schema(self) -> list[dict]:95        return self.get("schema")9697    # ------------------------------------------------------------ projets98    def projects(self, **filters) -> dict:99        """Une page : {total, limit, offset, items}. Filtres : q, type, sector, status, ticker, cik, location, tech,100        partner, min_amount, max_amount, year_from, year_to, min_confidence, has_amount, sort, order, limit, offset."""101        return self.get("projects", **filters)102103    def iter_projects(self, page_size: int = 500, max_items: int | None = None, **filters) -> Iterator[dict]:104        """Itère sur tous les projets correspondant aux filtres (pagination automatique)."""105        yield from self._paginate("projects", page_size, max_items, **filters)106107    def project(self, project_id: str) -> dict:108        """Fiche complète : {project, timeline, mentions, graph, similar}."""109        return self.get(f"projects/{project_id}")110111    def similar(self, project_id: str, k: int = 10) -> list[dict]:112        return self.get(f"projects/{project_id}/similar", k=k)113114    def projects_csv(self, **filters) -> str:115        return self.request("GET", "projects/export.csv", filters, raw=True)116117    # ------------------------------------------------------------ mentions / sections118    def mentions(self, **filters) -> dict:119        return self.get("mentions", **filters)120121    def iter_mentions(self, page_size: int = 500, max_items: int | None = None, **filters) -> Iterator[dict]:122        yield from self._paginate("mentions", page_size, max_items, **filters)123124    def mention(self, mention_id: str) -> dict:125        return self.get(f"mentions/{mention_id}")126127    def section(self, section_id: str, highlight: str | None = None) -> dict:128        return self.get(f"sections/{section_id}", highlight=highlight)129130    # ------------------------------------------------------------ entreprises131    def companies(self, **filters) -> dict:132        return self.get("companies", **filters)133134    def iter_companies(self, page_size: int = 500, **filters) -> Iterator[dict]:135        yield from self._paginate("companies", page_size, None, **filters)136137    def company(self, ticker_or_cik: str) -> dict:138        return self.get(f"companies/{ticker_or_cik}")139140    # ------------------------------------------------------------ graphe / recherche / SQL141    def graph_search(self, q: str, type: str | None = None, limit: int = 30) -> list[dict]:142        return self.get("graph/search", q=q, type=type, limit=limit)143144    def graph_node(self, node_id: str, limit: int = 200) -> dict:145        return self.get(f"graph/node/{urllib.parse.quote(node_id, safe=':')}", limit=limit)146147    def semantic(self, q: str, k: int = 20, type: str | None = None, sector: str | None = None) -> list[dict]:148        return self.get("search/semantic", q=q, k=k, type=type, sector=sector)["items"]149150    def sql(self, query: str, limit: int = 500) -> dict:151        """Requête de lecture DuckDB. Retour : {columns, rows, truncated, elapsed_ms}."""152        return self.request("POST", "sql", body={"sql": query, "limit": limit})153154    def sql_records(self, query: str, limit: int = 500) -> list[dict]:155        r = self.sql(query, limit)156        return [dict(zip(r["columns"], row)) for row in r["rows"]]157158    # ------------------------------------------------------------ utilitaires159    def _paginate(self, path: str, page_size: int, max_items: int | None, **filters) -> Iterator[dict]:160        offset, n = int(filters.pop("offset", 0) or 0), 0161        page_size = max(1, min(page_size, 500))162        while True:163            page = self.get(path, limit=page_size, offset=offset, **filters)164            items = page.get("items", [])165            for it in items:166                yield it167                n += 1168                if max_items and n >= max_items:169                    return170            offset += len(items)171            if not items or offset >= page.get("total", 0):172                return173174    @staticmethod175    def to_dataframe(records):176        """Convertit une liste/itérateur de dicts en DataFrame pandas (listes jointes par ' | ')."""177        import pandas as pd  # optionnel178        rows = []179        for r in records:180            rows.append({k: (" | ".join(map(str, v)) if isinstance(v, list) else v) for k, v in r.items()})181        return pd.DataFrame(rows)182183    @staticmethod184    def to_csv(records, path: str) -> int:185        rows = list(records)186        if not rows:187            return 0188        with open(path, "w", newline="", encoding="utf-8") as f:189            w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))190            w.writeheader()191            for r in rows:192                w.writerow({k: (" | ".join(map(str, v)) if isinstance(v, list) else v) for k, v in r.items()})193        return len(rows)194195196if __name__ == "__main__":  # petit test : python pdb_client.py VOTRE_CLE197    import sys198    c = PDB(sys.argv[1] if len(sys.argv) > 1 else None)199    print(json.dumps(c.health(), indent=2))200    ov = c.stats()["overview"]201    print(f"{ov['projects']:,} projets, {ov['mentions']:,} mentions, {ov['companies']} entreprises")202    for p in c.iter_projects(type="data_center", min_amount=5e8, sort="amount", max_items=5):203        print(f"  {p['ticker']:6s} {p['project_name'][:50]:50s} {p['total_amount_usd']/1e9:6.1f} G$")204