forked from 0xWheatyz/SPARC
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 0b4d712fc5 |
+11
-21
@@ -108,10 +108,12 @@ class CompanyAnalyzer:
|
|||||||
def analyze_single_patent(self, patent_id: str, company_name: str) -> str:
|
def analyze_single_patent(self, patent_id: str, company_name: str) -> str:
|
||||||
"""Analyze a single patent by ID.
|
"""Analyze a single patent by ID.
|
||||||
|
|
||||||
If the patent PDF is not already on disk, this method attempts to
|
Prerequisite:
|
||||||
download it automatically by looking up the PDF link in the database
|
The patent PDF must already exist at ``patents/{patent_id}.pdf``
|
||||||
cache. If the link is not cached either, a ``FileNotFoundError`` is
|
before calling this method. PDFs are downloaded automatically when
|
||||||
raised with instructions on how to obtain the PDF.
|
using the batch analysis pipeline (``analyze_company`` or the
|
||||||
|
``/analyze/batch`` API endpoint). For standalone usage, download
|
||||||
|
the PDF manually or call ``SERP.save_patents()`` first.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
patent_id: Publication ID of the patent (e.g. "US-11234567-B2")
|
patent_id: Publication ID of the patent (e.g. "US-11234567-B2")
|
||||||
@@ -121,7 +123,7 @@ class CompanyAnalyzer:
|
|||||||
Analysis of the specific patent's innovation quality
|
Analysis of the specific patent's innovation quality
|
||||||
|
|
||||||
Raises:
|
Raises:
|
||||||
FileNotFoundError: If the patent PDF cannot be found or downloaded.
|
FileNotFoundError: If the patent PDF is not found at the expected path.
|
||||||
"""
|
"""
|
||||||
import os
|
import os
|
||||||
logger.info("Analyzing patent %s for %s...", patent_id, company_name)
|
logger.info("Analyzing patent %s for %s...", patent_id, company_name)
|
||||||
@@ -129,22 +131,10 @@ class CompanyAnalyzer:
|
|||||||
patent_path = f"patents/{patent_id}.pdf"
|
patent_path = f"patents/{patent_id}.pdf"
|
||||||
|
|
||||||
if not os.path.exists(patent_path):
|
if not os.path.exists(patent_path):
|
||||||
# Attempt to download the PDF automatically from cached metadata
|
raise FileNotFoundError(
|
||||||
cached = self.db.get_cached_patent(patent_id)
|
f"Patent PDF not found at '{patent_path}'. "
|
||||||
pdf_link = cached.get("pdf_link") if cached else None
|
f"Download the PDF first using SERP.save_patents() or the batch analysis pipeline."
|
||||||
|
)
|
||||||
if pdf_link:
|
|
||||||
logger.info("PDF not on disk; downloading %s from cached link", patent_id)
|
|
||||||
patent = SERP.save_patents(
|
|
||||||
Patent(patent_id=patent_id, pdf_link=pdf_link)
|
|
||||||
)
|
|
||||||
patent_path = patent.pdf_path
|
|
||||||
else:
|
|
||||||
raise FileNotFoundError(
|
|
||||||
f"Patent PDF not found at '{patent_path}' and no download link is "
|
|
||||||
f"cached for '{patent_id}'. Run a company analysis first to populate "
|
|
||||||
f"the cache, or call SERP.save_patents() with the patent's PDF link."
|
|
||||||
)
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
sections = SERP.parse_patent_pdf(patent_path)
|
sections = SERP.parse_patent_pdf(patent_path)
|
||||||
|
|||||||
@@ -429,38 +429,6 @@ async def analyze_company(
|
|||||||
return _convert_result(result)
|
return _convert_result(result)
|
||||||
|
|
||||||
|
|
||||||
@app.get(
|
|
||||||
"/analyze/patent/{patent_id}",
|
|
||||||
tags=["Analysis"],
|
|
||||||
)
|
|
||||||
async def analyze_single_patent(
|
|
||||||
patent_id: str,
|
|
||||||
company_name: str = Query(description="Company name for analysis context"),
|
|
||||||
_: UserResponse = Depends(get_current_user),
|
|
||||||
):
|
|
||||||
"""Analyze a single patent by its publication ID.
|
|
||||||
|
|
||||||
If the patent PDF is not already cached locally, the system will attempt
|
|
||||||
to download it automatically from a previously cached link. If no link
|
|
||||||
is available, a 404 error is returned.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
patent_id: Patent publication ID (e.g. "US-11234567-B2")
|
|
||||||
company_name: Company name for analysis context
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Analysis text for the patent
|
|
||||||
"""
|
|
||||||
if not _analyzer:
|
|
||||||
raise HTTPException(status_code=503, detail="Analyzer not initialized")
|
|
||||||
|
|
||||||
try:
|
|
||||||
analysis = _analyzer.analyze_single_patent(patent_id, company_name)
|
|
||||||
return {"patent_id": patent_id, "company_name": company_name, "analysis": analysis}
|
|
||||||
except FileNotFoundError as e:
|
|
||||||
raise HTTPException(status_code=404, detail=str(e))
|
|
||||||
|
|
||||||
|
|
||||||
@app.post(
|
@app.post(
|
||||||
"/analyze/batch",
|
"/analyze/batch",
|
||||||
response_model=BatchAnalysisResponse,
|
response_model=BatchAnalysisResponse,
|
||||||
|
|||||||
+13
-1
@@ -1,3 +1,4 @@
|
|||||||
|
import logging
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
@@ -10,6 +11,8 @@ import serpapi
|
|||||||
from SPARC import config
|
from SPARC import config
|
||||||
from SPARC.types import Patent, Patents
|
from SPARC.types import Patent, Patents
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class SERP:
|
class SERP:
|
||||||
def query(company: str, days_back: int = None) -> Patents:
|
def query(company: str, days_back: int = None) -> Patents:
|
||||||
@@ -44,6 +47,7 @@ class SERP:
|
|||||||
"tbs": date_filter,
|
"tbs": date_filter,
|
||||||
"api_key": config.api_key,
|
"api_key": config.api_key,
|
||||||
}
|
}
|
||||||
|
logger.info("Querying Google Patents for '%s' (last %d days)", company, days_back)
|
||||||
search = serpapi.search(params)
|
search = serpapi.search(params)
|
||||||
# Convert results to Patent objects, skipping any without PDF links
|
# Convert results to Patent objects, skipping any without PDF links
|
||||||
patent_ids = []
|
patent_ids = []
|
||||||
@@ -52,8 +56,10 @@ class SERP:
|
|||||||
pdf_link = patent.get("pdf")
|
pdf_link = patent.get("pdf")
|
||||||
if pdf_link:
|
if pdf_link:
|
||||||
patent_ids.append(Patent(patent_id=patent["publication_number"], pdf_link=pdf_link, summary=None))
|
patent_ids.append(Patent(patent_id=patent["publication_number"], pdf_link=pdf_link, summary=None))
|
||||||
# Patents without PDF links are skipped (see docstring for details)
|
else:
|
||||||
|
logger.debug("Skipping patent %s (no PDF link)", patent.get("publication_number", "unknown"))
|
||||||
|
|
||||||
|
logger.info("Found %d patents with PDF links for '%s'", len(patent_ids), company)
|
||||||
return Patents(patents=patent_ids)
|
return Patents(patents=patent_ids)
|
||||||
|
|
||||||
def save_patents(patent: Patent) -> Patent:
|
def save_patents(patent: Patent) -> Patent:
|
||||||
@@ -70,9 +76,13 @@ class SERP:
|
|||||||
os.makedirs("patents", exist_ok=True)
|
os.makedirs("patents", exist_ok=True)
|
||||||
|
|
||||||
if not (os.path.exists(pdf_path) and os.path.getsize(pdf_path) > 0):
|
if not (os.path.exists(pdf_path) and os.path.getsize(pdf_path) > 0):
|
||||||
|
logger.info("Downloading PDF for %s", patent.patent_id)
|
||||||
response = requests.get(patent.pdf_link)
|
response = requests.get(patent.pdf_link)
|
||||||
with open(pdf_path, "wb") as f:
|
with open(pdf_path, "wb") as f:
|
||||||
f.write(response.content)
|
f.write(response.content)
|
||||||
|
logger.debug("Saved %d bytes to %s", len(response.content), pdf_path)
|
||||||
|
else:
|
||||||
|
logger.debug("Using cached PDF for %s at %s", patent.patent_id, pdf_path)
|
||||||
|
|
||||||
patent.pdf_path = pdf_path
|
patent.pdf_path = pdf_path
|
||||||
return patent
|
return patent
|
||||||
@@ -90,11 +100,13 @@ class SERP:
|
|||||||
Dictionary containing all extracted sections
|
Dictionary containing all extracted sections
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
logger.debug("Parsing patent PDF: %s", pdf_path)
|
||||||
with pdfplumber.open(pdf_path) as pdf:
|
with pdfplumber.open(pdf_path) as pdf:
|
||||||
# Extract all text
|
# Extract all text
|
||||||
full_text = ""
|
full_text = ""
|
||||||
for page in pdf.pages:
|
for page in pdf.pages:
|
||||||
full_text += page.extract_text() + "\n"
|
full_text += page.extract_text() + "\n"
|
||||||
|
logger.debug("Extracted text from %d pages (%d chars)", len(pdf.pages), len(full_text))
|
||||||
|
|
||||||
# Define section patterns (common in patents)
|
# Define section patterns (common in patents)
|
||||||
sections = {
|
sections = {
|
||||||
|
|||||||
Reference in New Issue
Block a user