diff --git a/app/api/v1/endpoints/filings.py b/app/api/v1/endpoints/filings.py index 8ead426..e4647d8 100644 --- a/app/api/v1/endpoints/filings.py +++ b/app/api/v1/endpoints/filings.py @@ -547,20 +547,26 @@ async def get_filing_events( response_model=BulkParseResponse, summary="Bulk parse pending 8-K filings", description=( - "Parse pending 8-K filings and create structured events. " - "Use this to backfill events for existing DB records.\\n\\n" - "**Example body**: `{\"tickers\": [\"AVGO\", \"AAPL\"], \"limit\": 50}`\\n" - "Omit `tickers` to parse all pending filings (up to `limit`)." + "Parse 8-K filings and create structured events.\\n\\n" + "- **Default**: processes only `pending` filings.\\n" + "- **`force_reparse=true`**: resets `succeeded`/`failed` filings to `pending` and re-parses them.\\n\\n" + "**Example — reparse specific ticker**: `{\"tickers\": [\"AVGO\"], \"limit\": 50, \"force_reparse\": true}`\\n" + "**Example — backfill all pending**: `{\"limit\": 200}`" ), ) async def parse_8k_bulk( request: BulkParseRequest, db: AsyncSession = Depends(get_db), ): - """Bulk parse pending 8-K filings and persist events.""" + """Bulk parse 8-K filings and persist events. Use force_reparse to retry failed/succeeded.""" try: result = await asyncio.wait_for( - sec_8k_parser.parse_bulk(db, tickers=request.tickers, limit=request.limit), + sec_8k_parser.parse_bulk( + db, + tickers=request.tickers, + limit=request.limit, + force_reparse=request.force_reparse, + ), timeout=600, ) except asyncio.TimeoutError: @@ -576,3 +582,50 @@ async def parse_8k_bulk( total=result["total"], query_time_seconds=result["elapsed"], ) + + +@router.post( + "/events/parse/{accession_number}", + response_model=BulkParseResponse, + summary="Force-reparse a single 8-K filing", + description=( + "Reparse a specific filing by accession number, regardless of current `parsed_status`.\\n\\n" + "**Example**: `POST /filings/events/parse/0001193125-26-144028`" + ), +) +async def parse_8k_single( + accession_number: str, + db: AsyncSession = Depends(get_db), +): + """Force-reparse a single filing (resets to pending, then parses).""" + import time as _time + t0 = _time.monotonic() + try: + # Reset status to pending regardless of current state + from app.models.filing import SECFiling as _SECFiling + await db.execute( + _SECFiling.__table__.update() + .where(_SECFiling.accession_number == accession_number) + .values(parsed_status="pending") + ) + await db.commit() + + n = await asyncio.wait_for( + sec_8k_parser.parse_filing(db, accession_number), + timeout=60, + ) + elapsed = round(_time.monotonic() - t0, 3) + return BulkParseResponse( + succeeded=1 if n > 0 else 0, + failed=0, + skipped=1 if n == 0 else 0, + total=1, + query_time_seconds=elapsed, + ) + except ValueError as e: + raise HTTPException(status_code=404, detail=str(e)) + except asyncio.TimeoutError: + raise HTTPException(status_code=504, detail="Parse timed out after 60s") + except Exception as e: + logger.error(f"Single parse failed for {accession_number}: {e}") + raise HTTPException(status_code=502, detail=f"Parse failed: {e}") diff --git a/app/schemas/filing.py b/app/schemas/filing.py index ec4f6a7..88ded7f 100644 --- a/app/schemas/filing.py +++ b/app/schemas/filing.py @@ -237,11 +237,15 @@ class FilingEventsSearchResponse(BaseModel): class BulkParseRequest(BaseModel): model_config = ConfigDict(json_schema_extra={ - "example": {"tickers": ["AVGO", "AAPL"], "limit": 50} + "example": {"tickers": ["AVGO", "AAPL"], "limit": 50, "force_reparse": False} }) tickers: Optional[List[str]] = None # None = all pending limit: int = Field(default=100, ge=1, le=1000) + force_reparse: bool = Field( + default=False, + description="If True, also reparse filings with status succeeded or failed (resets to pending first)", + ) class BulkParseResponse(BaseModel): diff --git a/app/services/sec_8k_parser.py b/app/services/sec_8k_parser.py index 9331786..0a92700 100644 --- a/app/services/sec_8k_parser.py +++ b/app/services/sec_8k_parser.py @@ -124,13 +124,31 @@ class SEC8KParser: db: AsyncSession, tickers: Optional[List[str]] = None, limit: int = 100, + force_reparse: bool = False, ) -> Dict: - """Parse pending 8-K filings, optionally filtered by ticker. + """Parse 8-K filings, optionally filtered by ticker. + Args: + force_reparse: If True, also reparse succeeded/failed filings (resets to pending). Returns summary dict with succeeded/failed/skipped counts. """ t0 = _time.monotonic() + # Reset failed/succeeded filings to pending when force_reparse is requested + if force_reparse: + reset_conditions = [ + SECFiling.parsed_status.in_(["succeeded", "failed"]), + SECFiling.form_type.in_(["8-K", "8-K/A"]), + ] + if tickers: + reset_conditions.append(SECFiling.ticker.in_([t.upper() for t in tickers])) + await db.execute( + SECFiling.__table__.update() + .where(and_(*reset_conditions)) + .values(parsed_status="pending") + ) + await db.commit() + conditions = [ SECFiling.parsed_status == "pending", SECFiling.form_type.in_(["8-K", "8-K/A"]),