bolade
341 lines
13 KiB
Python
341 lines
13 KiB
Python
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
# Import database models and compatibility score service
|
|
from db.models import InvestorTable, ProjectTable
|
|
from jinja2 import Environment, FileSystemLoader
|
|
from playwright.async_api import async_playwright
|
|
|
|
from services.compatibility_score import calculate_project_investor_compatibility
|
|
|
|
|
|
class ReportGenerator:
|
|
"""Service for generating PDF reports from HTML templates"""
|
|
|
|
def __init__(self):
|
|
# Set up Jinja2 environment
|
|
template_dir = Path(__file__).parent.parent / "templates"
|
|
self.env = Environment(loader=FileSystemLoader(str(template_dir)))
|
|
|
|
async def generate_investor_report(
|
|
self,
|
|
investor_data: Dict[str, Any],
|
|
project_data: Optional[Dict[str, Any]] = None,
|
|
investor_model: Optional[InvestorTable] = None,
|
|
project_model: Optional[ProjectTable] = None,
|
|
) -> bytes:
|
|
"""
|
|
Generate a PDF report for an investor profile.
|
|
|
|
Args:
|
|
investor_data: Dictionary containing investor information
|
|
project_data: Optional dictionary containing project information for compatibility analysis
|
|
investor_model: Optional database model for investor (used for compatibility scoring)
|
|
project_model: Optional database model for project (used for compatibility scoring)
|
|
|
|
Returns:
|
|
bytes: PDF file content
|
|
"""
|
|
# Prepare template context
|
|
context = self._prepare_context(
|
|
investor_data, project_data, investor_model, project_model
|
|
)
|
|
|
|
# Render HTML from template
|
|
template = self.env.get_template("report.html")
|
|
html_content = template.render(**context)
|
|
# Convert HTML to PDF using Playwright
|
|
pdf_bytes = await self._html_to_pdf(html_content)
|
|
|
|
return pdf_bytes
|
|
|
|
def _prepare_context(
|
|
self,
|
|
investor_data: Dict[str, Any],
|
|
project_data: Optional[Dict[str, Any]] = None,
|
|
investor_model: Optional[InvestorTable] = None,
|
|
project_model: Optional[ProjectTable] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Prepare the context dictionary for template rendering"""
|
|
context = {
|
|
"investor": investor_data,
|
|
"project": project_data,
|
|
"compatibility_score": 0,
|
|
"match_criteria": [],
|
|
"recommendation": None,
|
|
}
|
|
|
|
# If project data is provided, calculate compatibility
|
|
if project_data:
|
|
# Use the compatibility_score service if models are provided
|
|
if investor_model and project_model:
|
|
# Calculate using the standardized compatibility score service
|
|
# Returns score between 0 and 1, convert to percentage (0-100)
|
|
score_decimal = calculate_project_investor_compatibility(
|
|
project=project_model, investor=investor_model, use_funds=True
|
|
)
|
|
context["compatibility_score"] = int(score_decimal * 100)
|
|
else:
|
|
# Fallback to old calculation method if models not provided
|
|
context["compatibility_score"] = self._calculate_compatibility_score(
|
|
investor_data, project_data
|
|
)
|
|
|
|
context["match_criteria"] = self._generate_match_criteria(
|
|
investor_data, project_data
|
|
)
|
|
context["recommendation"] = self._generate_recommendation(
|
|
context["compatibility_score"], context["match_criteria"]
|
|
)
|
|
|
|
return context
|
|
|
|
def _calculate_compatibility_score(
|
|
self, investor_data: Dict[str, Any], project_data: Dict[str, Any]
|
|
) -> int:
|
|
"""Calculate overall compatibility score between investor and project"""
|
|
score = 0
|
|
weights = {
|
|
"sector": 30,
|
|
"stage": 30,
|
|
"geography": 20,
|
|
"check_size": 20,
|
|
}
|
|
|
|
# Aggregate data from all funds
|
|
all_sectors = set(investor_data.get("sectors", []))
|
|
all_stages = set()
|
|
all_geographies = []
|
|
check_ranges = []
|
|
|
|
for fund in investor_data.get("funds", []):
|
|
all_sectors.update(fund.get("sectors", []))
|
|
all_stages.update(fund.get("investment_stages", []))
|
|
if fund.get("geographic_focus"):
|
|
all_geographies.append(fund["geographic_focus"])
|
|
if fund.get("check_size_lower") and fund.get("check_size_upper"):
|
|
check_ranges.append(
|
|
{
|
|
"lower": fund["check_size_lower"],
|
|
"upper": fund["check_size_upper"],
|
|
}
|
|
)
|
|
|
|
# Sector match
|
|
project_sectors = set(project_data.get("sectors", []))
|
|
if all_sectors and project_sectors:
|
|
if all_sectors & project_sectors:
|
|
score += weights["sector"]
|
|
|
|
# Stage match - case insensitive comparison
|
|
project_stage = project_data.get("stage")
|
|
if project_stage and all_stages:
|
|
# Normalize stage names for comparison (case-insensitive)
|
|
normalized_stages = {
|
|
stage.lower().replace("_", " ") for stage in all_stages
|
|
}
|
|
project_stage_normalized = project_stage.lower().replace("_", " ")
|
|
if project_stage_normalized in normalized_stages:
|
|
score += weights["stage"]
|
|
|
|
# Geography match - check if any fund matches
|
|
project_geo = (project_data.get("location") or "").lower()
|
|
geo_match = False
|
|
if all_geographies:
|
|
for geo in all_geographies:
|
|
if geo:
|
|
geo_lower = geo.lower()
|
|
# Match if investor geography is "global" or if there's a location overlap
|
|
if "global" in geo_lower or "worldwide" in geo_lower:
|
|
geo_match = True
|
|
break
|
|
if project_geo and (
|
|
geo_lower in project_geo or project_geo in geo_lower
|
|
):
|
|
geo_match = True
|
|
break
|
|
if geo_match:
|
|
score += weights["geography"]
|
|
|
|
# Check size match - check if any fund's range matches
|
|
project_valuation = project_data.get("valuation", 0)
|
|
check_match = False
|
|
if project_valuation and check_ranges:
|
|
for check_range in check_ranges:
|
|
if check_range["lower"] <= project_valuation <= check_range["upper"]:
|
|
check_match = True
|
|
break
|
|
if check_match:
|
|
score += weights["check_size"]
|
|
|
|
return min(score, 100)
|
|
|
|
def _generate_match_criteria(
|
|
self, investor_data: Dict[str, Any], project_data: Dict[str, Any]
|
|
) -> List[Dict[str, str]]:
|
|
"""Generate detailed match criteria table"""
|
|
criteria = []
|
|
|
|
# Aggregate data from all funds
|
|
all_sectors = set(investor_data.get("sectors", []))
|
|
all_stages = set()
|
|
all_geographies = []
|
|
check_ranges = []
|
|
|
|
for fund in investor_data.get("funds", []):
|
|
all_sectors.update(fund.get("sectors", []))
|
|
all_stages.update(fund.get("investment_stages", []))
|
|
if fund.get("geographic_focus"):
|
|
all_geographies.append(fund["geographic_focus"])
|
|
if fund.get("check_size_lower") and fund.get("check_size_upper"):
|
|
check_ranges.append(
|
|
{
|
|
"lower": fund["check_size_lower"],
|
|
"upper": fund["check_size_upper"],
|
|
"fund_name": fund.get("fund_name", "Unnamed Fund"),
|
|
}
|
|
)
|
|
|
|
# Sector criterion
|
|
project_sectors = project_data.get("sectors", [])
|
|
sector_match = "Perfect" if all_sectors & set(project_sectors) else "Mismatch"
|
|
criteria.append(
|
|
{
|
|
"name": "Sector",
|
|
"requirement": ", ".join(project_sectors) if project_sectors else "N/A",
|
|
"evidence": ", ".join(list(all_sectors)[:3]) if all_sectors else "N/A",
|
|
"match": sector_match,
|
|
"weight": "30%",
|
|
}
|
|
)
|
|
|
|
# Stage criterion - case insensitive comparison
|
|
project_stage = project_data.get("stage", "N/A")
|
|
stage_match = "Mismatch"
|
|
if project_stage != "N/A" and all_stages:
|
|
# Normalize stage names for comparison
|
|
normalized_stages = {
|
|
stage.lower().replace("_", " ") for stage in all_stages
|
|
}
|
|
project_stage_normalized = project_stage.lower().replace("_", " ")
|
|
stage_match = (
|
|
"Perfect"
|
|
if project_stage_normalized in normalized_stages
|
|
else "Mismatch"
|
|
)
|
|
elif project_stage == "N/A":
|
|
stage_match = "N/A"
|
|
|
|
criteria.append(
|
|
{
|
|
"name": "Stage",
|
|
"requirement": str(project_stage),
|
|
"evidence": ", ".join(all_stages) if all_stages else "N/A",
|
|
"match": stage_match,
|
|
"weight": "30%",
|
|
}
|
|
)
|
|
|
|
# Geography criterion
|
|
project_geo = project_data.get("location") or "N/A"
|
|
investor_geo_display = ", ".join(all_geographies) if all_geographies else "N/A"
|
|
|
|
# Safe comparison handling None values and "Global" matches
|
|
geo_match = "Mismatch"
|
|
if project_geo != "N/A" and all_geographies:
|
|
for geo in all_geographies:
|
|
if geo:
|
|
geo_lower = geo.lower()
|
|
# Match if investor geography is "global" or if there's a location overlap
|
|
if "global" in geo_lower or "worldwide" in geo_lower:
|
|
geo_match = "Perfect"
|
|
break
|
|
if (
|
|
geo_lower in project_geo.lower()
|
|
or project_geo.lower() in geo_lower
|
|
):
|
|
geo_match = "Strong"
|
|
break
|
|
elif not all_geographies and project_geo == "N/A":
|
|
geo_match = "N/A"
|
|
|
|
criteria.append(
|
|
{
|
|
"name": "Geography",
|
|
"requirement": project_geo,
|
|
"evidence": investor_geo_display,
|
|
"match": geo_match,
|
|
"weight": "20%",
|
|
}
|
|
)
|
|
|
|
# Check Size criterion
|
|
project_val = project_data.get("valuation", 0)
|
|
|
|
# Build evidence string from all fund ranges
|
|
check_evidence = "N/A"
|
|
if check_ranges:
|
|
evidence_parts = []
|
|
for cr in check_ranges[:3]: # Show up to 3 funds
|
|
range_str = (
|
|
f"€{cr['lower'] / 1000000:.0f}M - €{cr['upper'] / 1000000:.0f}M"
|
|
)
|
|
if cr["fund_name"]:
|
|
evidence_parts.append(f"{cr['fund_name']}: {range_str}")
|
|
else:
|
|
evidence_parts.append(range_str)
|
|
check_evidence = "; ".join(evidence_parts)
|
|
|
|
# Check if project valuation matches any fund
|
|
check_match = "N/A"
|
|
if project_val > 0 and check_ranges:
|
|
match_found = any(
|
|
cr["lower"] <= project_val <= cr["upper"] for cr in check_ranges
|
|
)
|
|
check_match = "Perfect" if match_found else "Mismatch"
|
|
|
|
criteria.append(
|
|
{
|
|
"name": "Check Size",
|
|
"requirement": f"€{project_val / 1000000:.0f}M"
|
|
if project_val
|
|
else "N/A",
|
|
"evidence": check_evidence,
|
|
"match": check_match,
|
|
"weight": "20%",
|
|
}
|
|
)
|
|
|
|
return criteria
|
|
|
|
def _generate_recommendation(
|
|
self, score: int, criteria: List[Dict[str, str]]
|
|
) -> str:
|
|
"""Generate recommendation text based on score and criteria"""
|
|
if score >= 85:
|
|
return "High Priority. A strong target due to exceptional alignment on the most heavily-weighted criteria: Sector and Stage. The strong geographic fit further solidifies this recommendation."
|
|
elif score >= 70:
|
|
return "Medium Priority. Good alignment on key criteria with some areas of strong fit. The geographic fit in the target region supports this recommendation."
|
|
else:
|
|
return "Low Priority. Limited alignment on key investment criteria. Consider for future evaluation if circumstances change."
|
|
|
|
async def _html_to_pdf(self, html_content: str) -> bytes:
|
|
"""Convert HTML content to PDF using Playwright"""
|
|
async with async_playwright() as p:
|
|
browser = await p.chromium.launch()
|
|
page = await browser.new_page()
|
|
|
|
# Set content and wait for any dynamic content to load
|
|
await page.set_content(html_content, wait_until="networkidle")
|
|
|
|
# Generate PDF with proper settings
|
|
pdf_bytes = await page.pdf(
|
|
format="A4",
|
|
print_background=True,
|
|
margin={"top": "0", "right": "0", "bottom": "0", "left": "0"},
|
|
)
|
|
|
|
await browser.close()
|
|
|
|
return pdf_bytes
|