import os import asyncio import logging from typing import List, Dict, Any, Optional import re import time import base64 import io import tempfile # CRITICAL FIX: Configure matplotlib backend BEFORE importing matplotlib import matplotlib matplotlib.use('Agg') # Use non-interactive backend for server environments import matplotlib.pyplot as plt import numpy as np from dotenv import load_dotenv load_dotenv() # Load environment variables from .env into os.environ from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi import Query from fastapi.responses import FileResponse from pydantic import BaseModel from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials # Import LangChain components from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain.schema.runnable import RunnableSequence from langchain.schema import AIMessage # Import the base LLM class to build our custom wrapper from langchain.llms.base import LLM from huggingface_hub import InferenceClient import jwt as PyJWT import asyncpg import uuid from contextlib import asynccontextmanager # Set up logging logging.basicConfig(level=logging.INFO) # Database connection pool DB_POOL = None def get_db_pool(): """Get the database connection pool""" global DB_POOL return DB_POOL def set_db_pool(pool): """Set the database connection pool""" global DB_POOL DB_POOL = pool return DB_POOL # JWT verification function def get_user_id_from_request(request: Request) -> Optional[str]: """Extract and verify user ID from JWT token in request headers""" try: auth_header = request.headers.get("Authorization") if not auth_header or not auth_header.startswith("Bearer "): return None token = auth_header.split(" ")[1] # Supabase JWT verification SUPABASE_JWT_SECRET = os.getenv("SUPABASE_JWT_SECRET") SUPABASE_JWT_AUD = os.getenv("SUPABASE_JWT_AUD") # optional; e.g. "authenticated" if not token or not SUPABASE_JWT_SECRET: return None try: if SUPABASE_JWT_AUD: payload = PyJWT.decode(token, SUPABASE_JWT_SECRET, algorithms=["HS256"], audience=SUPABASE_JWT_AUD) else: payload = PyJWT.decode(token, SUPABASE_JWT_SECRET, algorithms=["HS256"]) return payload.get("sub") # user ID from JWT except PyJWT.InvalidTokenError: return None except Exception as e: logging.warning(f"JWT verification failed: {e}") return None # Check for required environment variables def check_environment_variables(): hf_token = os.getenv("HUGGINGFACEHUB_API_TOKEN") if not hf_token: logging.warning("⚠️ HUGGINGFACEHUB_API_TOKEN not set. Hugging Face models will not work properly.") else: logging.info("✅ HUGGINGFACEHUB_API_TOKEN is set") openai_api_key = os.getenv("OPENAI_API_KEY") if not openai_api_key: logging.warning("⚠️ OPENAI_API_KEY not set. OpenAI GPT models will not work properly.") else: logging.info("✅ OPENAI_API_KEY is set") # Check Supabase configuration supabase_url = os.getenv("DATABASE_URL") supabase_key = os.getenv("SUPABASE_JWT_SECRET") if not supabase_url or not supabase_key: logging.warning("⚠️ DATABASE_URL or SUPABASE_KEY not set. Database persistence will not work.") else: logging.info("✅ Supabase configuration is set") # Check JWT configuration jwt_secret = os.getenv("SUPABASE_JWT_SECRET") if not jwt_secret: logging.warning("⚠️ SUPABASE_JWT_SECRET not set. JWT verification will not work.") else: logging.info("✅ JWT configuration is set") # Note: UploadThing is now handled by frontend, not server-side logging.info("ℹ️ UploadThing uploads are handled by frontend using document data from server") # Run the environment variable check check_environment_variables() # Display information about model size limitations logging.info("=" * 80) logging.info("MODEL SIZE LIMITATIONS:") logging.info("The free tier of Hugging Face Inference API limits models to 10GB.") logging.info("Large models like Qwen-2.5-7B (15GB) and Llama-2-7B (13GB) exceed this limit.") logging.info("We've configured smaller alternative models as replacements.") logging.info("For full-sized models, upgrade to Hugging Face Pro subscription.") logging.info("=" * 80) # Supabase connection setup SUPABASE_URL = os.getenv("DATABASE_URL") SUPABASE_KEY = os.getenv("SUPABASE_JWT_SECRET") # JWT security security = HTTPBearer() # Create the FastAPI app app = FastAPI() # Add logging middleware @app.middleware("http") async def log_requests(request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time logging.info(f"Request: {request.method} {request.url.path} - Status: {response.status_code} - Time: {process_time:.2f}s") return response # Configure CORS for Hugging Face Spaces app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], expose_headers=["*"] ) # Allow all origins (adjust for production usage) # app.add_middleware( # CORSMiddleware, # allow_origins=["*"], # allow_credentials=True, # allow_methods=["*"], # allow_headers=["*"], # ) # Fixed list of business questions (order matters) QUESTIONS = [ "What is your business name?", "What product or service do you offer?", "Who is your target customer?", "What problem does your business solve?", "Who are your competitors?", "What is your unique value proposition?", "What is your pricing strategy?", "What are your short-term goals?", "What are your long-term goals?", "How will you acquire customers?", ] # Data model for incoming business plan generation requests class GenerateRequest(BaseModel): answers: List[str] model: str prompt: str # The prompt template from the database promptVariables: Dict[str, str] # Variables to fill in the template provider: str # The provider (e.g., "hf-inference", "together") temperature: float # Temperature setting maxTokens: int # Max tokens setting # NEW — optional, for persistence & ownership businessPlanId: Optional[str] = None planType: Optional[str] = "free" userId: Optional[str] = None # will be overridden by verified JWT anonymousId: Optional[str] = None countryCode: Optional[str] = None sectionKey: Optional[str] = None # e.g. "prompt_ExecutiveSummary" # Data models for HF Export functionality class ExportSection(BaseModel): key: str title: str content: str class ExportRequest(BaseModel): summary: str sections: List[ExportSection] businessIdea: Optional[str] = None isFullPlan: Optional[bool] = False format: str # "pdf" or "word" chartImages: Optional[List[str]] = None # Base64 encoded images class ExportResponse(BaseModel): success: bool downloadUrl: str filename: str size: int message: str documentData: Optional[str] = None # Base64 encoded document data for frontend upload # Database helper constants SECTION_MAP = { "prompt_ExecutiveSummary": ("executiveSummary", "executiveSummaryComplete", "executiveSummaryGenerating"), "prompt_CompanyProfile": ("companyProfile", "companyProfileComplete", "companyProfileGenerating"), "prompt_MarketAnalysis": ("marketAnalysis", "marketAnalysisComplete", "marketAnalysisGenerating"), "prompt_ProductOrService": ("productOrService", "productOrServiceComplete", "productOrServiceGenerating"), "prompt_BusinessModel": ("businessModel", "businessModelComplete", "businessModelGenerating"), "prompt_MarketingGrowth": ("marketingGrowth", "marketingGrowthComplete", "marketingGrowthGenerating"), "prompt_OperationsPlan": ("operationsPlan", "operationsPlanComplete", "operationsPlanGenerating"), "prompt_ManagementTeam": ("managementTeam", "managementTeamComplete", "managementTeamGenerating"), "prompt_FinancialPlanFunding": ("financialPlanFunding", "financialPlanFundingComplete", "financialPlanFundingGenerating"), } SECTION_ORDER = [ "Executive Summary", "Company Profile", "Market Analysis", "Product or Service", "Business Model", "Marketing & Growth", "Operations Plan", "Management Team", "Financial Plan & Funding" ] def sort_sections_by_order(sections: List[ExportSection]) -> List[ExportSection]: """Sort sections according to the predefined SECTION_ORDER to ensure correct numerical ordering""" # Create a mapping from section title to order index order_map = {title: index for index, title in enumerate(SECTION_ORDER)} def get_section_order(section: ExportSection) -> int: # Try to match by exact title first if section.title in order_map: order_index = order_map[section.title] return order_index # Try case-insensitive matching section_title_lower = section.title.lower() for order_title, order_index in order_map.items(): if order_title.lower() == section_title_lower: return order_index # Try to match by key (remove 'prompt_' prefix if present) key_clean = section.key.replace('prompt_', '') if section.key.startswith('prompt_') else section.key if key_clean in order_map: order_index = order_map[key_clean] return order_index # Try partial matching for keys for order_title, order_index in order_map.items(): if key_clean.lower() in order_title.lower() or order_title.lower() in key_clean.lower(): return order_index # If no match found, put at the end return len(SECTION_ORDER) # Sort sections by their order sorted_sections = sorted(sections, key=get_section_order) return sorted_sections def extract_summary_from_markdown(markdown_text: str, max_length: int = 200) -> str: """Extract a summary from markdown text, removing headers and formatting""" if not markdown_text: return "" # Try to find the first header as a title lines = markdown_text.split('\n') for line in lines: if line.strip().startswith('#'): header_text = re.sub(r'^#+\s*', '', line.strip()) if header_text: return header_text[:max_length] # Fallback to first non-empty line for line in lines: clean_line = re.sub(r'[*_`~#]', '', line.strip()) if clean_line: return clean_line[:max_length] return "Generated Business Plan" async def _ensure_business_plan(pool, *, bp_id: Optional[str], summary: str, full_plan: str, model: str, answers: List[str], plan_type: Optional[str], user_id: Optional[str], anon_id: Optional[str], country_code: Optional[str]) -> str: """Ensure a business plan exists and return its ID""" async with pool.acquire() as conn: if bp_id: exists = await conn.fetchval('SELECT 1 FROM "BusinessPlan" WHERE id = $1', bp_id) if exists: await conn.execute( 'UPDATE "BusinessPlan" SET ' '"summary"=$1, "fullPlan"=$2, "model"=$3, "answers"=$4, ' '"planType"=COALESCE($5,\'free\'), "countryCode"=$6, ' '"updatedAt"=now(), "userId"=COALESCE($7,"userId") ' 'WHERE id=$8', summary, full_plan, model, answers, plan_type, country_code, user_id, bp_id ) return bp_id new_id = str(uuid.uuid4()) row = await conn.fetchrow( 'INSERT INTO "BusinessPlan" ' '("id","summary","fullPlan","model","answers","planType","userId","anonymousId","countryCode","updatedAt") ' 'VALUES ($1,$2,$3,$4,$5,COALESCE($6,\'free\'),$7,$8,$9, now()) ' 'RETURNING id', new_id, summary, full_plan, model, answers, plan_type, user_id, anon_id, country_code ) return row["id"] async def _upsert_section(pool, *, business_plan_id: str, section_key: str, content: str) -> None: """Update or insert a business plan section""" if section_key not in SECTION_MAP: logging.warning(f"Unknown section_key '{section_key}', skipping section save.") return col_content, col_complete, col_generating = SECTION_MAP[section_key] set_clause = f'"{col_content}" = $2, "{col_complete}" = TRUE, "{col_generating}" = FALSE' async with pool.acquire() as conn: # ensure row exists with a generated id await conn.execute( 'INSERT INTO "BusinessPlanSection" ("id","businessPlanId") ' 'VALUES ($1,$2) ' 'ON CONFLICT ("businessPlanId") DO NOTHING', str(uuid.uuid4()), business_plan_id ) # update content + flags await conn.execute( f'UPDATE "BusinessPlanSection" SET {set_clause} WHERE "businessPlanId" = $1', business_plan_id, content ) async def _recompute_full_plan(pool, *, business_plan_id: str) -> str: """Recompute the full plan from all sections""" async with pool.acquire() as conn: row = await conn.fetchrow( 'SELECT * FROM "BusinessPlanSection" WHERE "businessPlanId" = $1', business_plan_id ) if not row: return "" parts = [] for col in SECTION_ORDER: txt = row.get(col) if txt: parts.append(txt.strip()) full_plan = "\n\n".join(parts) await conn.execute( 'UPDATE "BusinessPlan" SET "fullPlan"=$1, "updatedAt"=now() WHERE id=$2', full_plan, business_plan_id ) return full_plan def _all_sections_present(row: asyncpg.Record) -> bool: """Check if all required sections are present""" return all(bool(row.get(col)) for col in SECTION_ORDER) def clean_markdown_text(text: str) -> str: """Clean markdown formatting for PDF generation - PRESERVE HEADINGS""" if not text: return "" # Remove markdown formatting BUT PRESERVE HEADING STRUCTURE cleaned = text # PRESERVE HEADINGS - don't strip the # markers, keep them for rendering # We'll handle styling at render time instead of stripping them # Remove bold, italic, code, strikethrough cleaned = re.sub(r'\*\*(.*?)\*\*', r'\1', cleaned) # Remove bold cleaned = re.sub(r'\*(.*?)\*', r'\1', cleaned) # Remove italic cleaned = re.sub(r'`(.*?)`', r'\1', cleaned) # Remove code cleaned = re.sub(r'~~(.*?)~~', r'\1', cleaned) # Remove strikethrough # Remove links (keep text) cleaned = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', cleaned) # Remove links, keep text cleaned = re.sub(r'!\[([^\]]*)\]\([^)]+\)', '', cleaned) # Remove images # Remove list markers (more aggressive) cleaned = re.sub(r'^\s*[-*+]\s*', '', cleaned, flags=re.MULTILINE) # Remove list markers cleaned = re.sub(r'^\s*\d+\.\s*', '', cleaned, flags=re.MULTILINE) # Remove numbered list markers cleaned = re.sub(r'^\s*[-*+]\s*', '', cleaned, flags=re.MULTILINE) # Remove any remaining list markers # Remove blockquotes cleaned = re.sub(r'^\s*>\s*', '', cleaned, flags=re.MULTILINE) # Remove blockquotes # Remove horizontal rules cleaned = re.sub(r'^\s*[-*_]{3,}\s*$', '', cleaned, flags=re.MULTILINE) # Remove horizontal rules # Remove code blocks cleaned = re.sub(r'```[\s\S]*?```', '', cleaned) # Remove code blocks cleaned = re.sub(r'`.*?`', '', cleaned) # Remove any remaining inline code # Remove emphasis markers cleaned = re.sub(r'_{1,2}(.*?)_{1,2}', r'\1', cleaned) # Remove underscores cleaned = re.sub(r'\*{1,2}(.*?)\*{1,2}', r'\1', cleaned) # Remove asterisks # REMOVE RAW CHART DATA MARKERS (CRITICAL FIX) cleaned = re.sub(r'.*?', '', cleaned, flags=re.DOTALL) cleaned = re.sub(r'.*$', '', cleaned, flags=re.DOTALL) cleaned = re.sub(r'', '', cleaned) # Remove any remaining HTML-like tags cleaned = re.sub(r'<[^>]+>', '', cleaned) # Clean up extra whitespace and formatting cleaned = re.sub(r'\n\s*\n\s*\n+', '\n\n', cleaned) # Remove excessive line breaks cleaned = re.sub(r'^\s+', '', cleaned, flags=re.MULTILINE) # Remove leading whitespace cleaned = re.sub(r'\s+$', '', cleaned, flags=re.MULTILINE) # Remove trailing whitespace cleaned = re.sub(r' +', ' ', cleaned) # Replace multiple spaces with single space # Final cleanup cleaned = cleaned.strip() return cleaned def clean_text_for_pdf(text: str) -> str: """Clean text specifically for PDF generation, handling Unicode issues""" if not text: return "" # First clean markdown cleaned = clean_markdown_text(text) # Replace problematic Unicode characters with ASCII equivalents unicode_replacements = { '\u2019': "'", # Right single quotation mark '\u2018': "'", # Left single quotation mark '\u201C': '"', # Left double quotation mark '\u201D': '"', # Right double quotation mark '\u2013': '-', # En dash '\u2014': '--', # Em dash '\u2022': '•', # Bullet '\u2026': '...', # Horizontal ellipsis '\u00A0': ' ', # Non-breaking space '\u00B0': '°', # Degree sign '\u00AE': '(R)', # Registered trademark '\u2122': '(TM)', # Trademark '\u00A9': '(C)', # Copyright } for unicode_char, replacement in unicode_replacements.items(): cleaned = cleaned.replace(unicode_char, replacement) return cleaned def generate_model_output(model: str, provider: str, api_key: str, prompt: str, max_tokens: int = 4000) -> str: """ A helper function that wraps the Hugging Face Inference API call. """ try: logging.info(f"Initializing InferenceClient with provider: {provider}") client = InferenceClient(provider=provider, api_key=api_key) logging.info(f"Sending request to model: {model} with prompt length: {len(prompt)} and max_tokens: {max_tokens}") completion = client.chat.completions.create( model=model, messages=[{"role": "user", "content": prompt}], max_tokens=max_tokens, ) logging.info(f"Successfully received response from model: {model} with content length: {len(completion.choices[0].message.content)}") return completion.choices[0].message.content except Exception as e: error_message = f"Error generating output with model {model}: {str(e)}" logging.error(error_message) # Re-raise the exception with more context raise Exception(error_message) from e ############################################################################### # HFInferenceLLM: A wrapper for Hugging Face models that matches the LLM interface. ############################################################################### class HFInferenceLLM(LLM): model: str = None provider: str = "hf-inference" api_key: str = "" max_tokens: int = 4000 # Increased from 500 to 4000 by default def __init__(self, model: str, provider: str = "hf-inference", api_key: str = "", max_tokens: int = 4000): super().__init__() self.model = model self.provider = provider self.api_key = api_key self.max_tokens = max_tokens # Validate API key if not api_key: logging.error(f"No API key provided for model {model}") raise ValueError(f"API key is required for Hugging Face Inference API access to {model}") # Check if model is available try: # Create a client to validate connection client = InferenceClient(provider=provider, api_key=api_key) logging.info(f"Successfully initialized client for model: {model}") except Exception as e: logging.error(f"Failed to initialize client for model {model}: {str(e)}") raise ValueError(f"Could not initialize Hugging Face client for {model}: {str(e)}") @property def _llm_type(self) -> str: return "hf_inference" def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str: """ Calls the Hugging Face API using the provided prompt. The `stop` parameter is not implemented (or forwarded) here. """ return generate_model_output( model=self.model, provider=self.provider, api_key=self.api_key, prompt=prompt, max_tokens=self.max_tokens ) def get_num_tokens(self, prompt: str) -> int: """ A simple implementation that counts tokens as space-separated words. You may replace this with a more accurate token counter. """ return len(prompt.split()) ############################################################################### # get_llm: Return an LLM instance based on the selected model. ############################################################################### def get_llm(model_name: str, provider: str = "hf-inference"): hf_token = os.getenv("HUGGINGFACEHUB_API_TOKEN") if not hf_token: logging.error("HUGGINGFACEHUB_API_TOKEN environment variable not set") raise ValueError("HUGGINGFACEHUB_API_TOKEN environment variable is required for Hugging Face models") if model_name == "BPGenerateAI": openai_api_key = os.getenv("OPENAI_API_KEY") if not openai_api_key: logging.error("OPENAI_API_KEY environment variable not set") raise ValueError("OPENAI_API_KEY environment variable is required for GPT models") return ChatOpenAI( model_name="gpt-4.1-mini", temperature=0.7, max_tokens=6000, openai_api_key=openai_api_key ) elif model_name == "BPSuggestionsAI": openai_api_key = os.getenv("OPENAI_API_KEY") if not openai_api_key: logging.error("OPENAI_API_KEY environment variable not set") raise ValueError("OPENAI_API_KEY environment variable is required for GPT models") return ChatOpenAI( model_name="gpt-4.1-nano", temperature=0.7, max_tokens=4000, openai_api_key=openai_api_key ) elif model_name.lower() == "gpt-4.1-nano": openai_api_key = os.getenv("OPENAI_API_KEY") if not openai_api_key: raise ValueError("OPENAI_API_KEY is required for GPT models") return ChatOpenAI( model_name="gpt-4.1-nano", temperature=0.7, max_tokens=4000, openai_api_key=openai_api_key ) else: # For all other models, use the specified provider return HFInferenceLLM( model=model_name, provider=provider, api_key=hf_token, max_tokens=4000 ) ############################################################################### # FastAPI Endpoints ############################################################################### @app.get("/suggestions", response_model=Dict[str, Any]) async def get_suggestions( business_idea: str, model: str, question: str, prompt: str, provider: str, temperature: float, maxTokens: int, exclude: List[str] = Query([]) ) -> Dict[str, Any]: try: # Use the prompt from the frontend formatted_prompt = prompt.format( business_idea=business_idea, question=question, exclude=", ".join(exclude) if exclude else "" ) # Get the LLM instance with frontend-provided configuration llm = get_llm(model, provider) if hasattr(llm, 'temperature'): llm.temperature = temperature if hasattr(llm, 'max_tokens'): llm.max_tokens = maxTokens suggestion_prompt = PromptTemplate( input_variables=[], template=formatted_prompt ) suggestion_chain: RunnableSequence = suggestion_prompt | llm raw_result = await asyncio.to_thread(suggestion_chain.invoke, {}) # Normalize AIMessage or list[AIMessage] → plain string if isinstance(raw_result, AIMessage): raw_text = raw_result.content elif ( isinstance(raw_result, (list, tuple)) and raw_result and isinstance(raw_result[0], AIMessage) ): raw_text = "\n".join(msg.content for msg in raw_result) else: raw_text = str(raw_result) # Clean up the suggestions suggestion_array = [ re.sub(r'^\s*[\-\d\.\)\s]+', '', line).strip() for line in raw_text.split("\n") if line.strip() ] if not suggestion_array: suggestion_array = ["No suggestions available"] return {"suggestions": suggestion_array} except Exception as e: raise HTTPException(status_code=500, detail=f"Error generating suggestions: {str(e)}") @app.post("/generate") async def generate_business_plan(request: Request, data: GenerateRequest): """ Generate a business plan using the provided prompt template and variables. """ try: # Get the LLM instance based on the selected model and provider logging.info(f"Initializing model: {data.model} with provider: {data.provider}") llm_selected = get_llm(data.model) # Set the temperature and max tokens from the request if hasattr(llm_selected, 'temperature'): llm_selected.temperature = data.temperature if hasattr(llm_selected, 'max_tokens'): llm_selected.max_tokens = data.maxTokens # Create the prompt template plan_prompt = PromptTemplate( input_variables=list(data.promptVariables.keys()), template=data.prompt ) # Create the chain plan_chain: RunnableSequence = plan_prompt | llm_selected logging.info(f"Generating business plan with model: {data.model}") try: # Invoke the chain with the variables raw_plan = await asyncio.to_thread( plan_chain.invoke, data.promptVariables ) # ── NEW: derive userId from JWT (ignore client-sent userId) verified_user_id = get_user_id_from_request(request) if verified_user_id: data.userId = verified_user_id # enforce verified owner # else keep anonymousId path if provided # Unwrap AIMessage or list of AIMessages into a single markdown string if isinstance(raw_plan, AIMessage): plan_text = raw_plan.content elif isinstance(raw_plan, (list, tuple)) and raw_plan and isinstance(raw_plan[0], AIMessage): plan_text = "\n".join(msg.content for msg in raw_plan) else: plan_text = str(raw_plan) # Extract summary from the generated plan summary = extract_summary_from_markdown(plan_text) # ── Persist to database if pool is available pool = getattr(app.state, "db_pool", None) if not pool: logging.warning("DB pool not available; skipping persistence.") return { "summary": summary or "Generated Business Plan", "plan": plan_text, } # If generating a specific section if data.sectionKey: business_plan_id = await _ensure_business_plan( pool, bp_id=data.businessPlanId, summary=summary, full_plan="", # recomputed after section save model=data.model, answers=data.answers, plan_type=data.planType, user_id=data.userId, anon_id=data.anonymousId, country_code=data.countryCode, ) await _upsert_section( pool, business_plan_id=business_plan_id, section_key=data.sectionKey, content=plan_text ) # Recompute full plan after section write full_plan = await _recompute_full_plan(pool, business_plan_id=business_plan_id) # Optional: mark complete if all sections present is_complete = False async with pool.acquire() as conn: row = await conn.fetchrow( 'SELECT * FROM "BusinessPlanSection" WHERE "businessPlanId"=$1', business_plan_id ) if row: is_complete = _all_sections_present(row) return { "summary": summary, "plan": plan_text, "businessPlanId": business_plan_id, "sections": [{"key": data.sectionKey, "content": plan_text}], "isComplete": is_complete } # Non-section — treat as a whole plan business_plan_id = await _ensure_business_plan( pool, bp_id=data.businessPlanId, summary=summary, full_plan=plan_text, model=data.model, answers=data.answers, plan_type=data.planType, user_id=data.userId, anon_id=data.anonymousId, country_code=data.countryCode, ) return { "summary": summary, "plan": plan_text, "businessPlanId": business_plan_id } except Exception as e: error_message = f"Error generating business plan with model {data.model}: {str(e)}" logging.error(error_message) raise HTTPException(status_code=500, detail=error_message) except Exception as e: error_message = f"Error initializing model {data.model}: {str(e)}" logging.error(error_message) raise HTTPException(status_code=500, detail=error_message) def create_pdf_document(request: ExportRequest): """Create a PROFESSIONAL PDF document from the business plan data using FPDF2 NOTE: Now uses matplotlib charts exclusively - old base64 chart logic commented out""" try: from fpdf import FPDF except ImportError: raise Exception("fpdf2 is not installed. Please install it with: pip install fpdf2") # Filter out empty sections to prevent blank pages filtered_sections = filter_empty_sections(request.sections) if not filtered_sections: raise Exception("No valid sections found after filtering. All sections appear to be empty.") # Generate table of contents toc_items = generate_table_of_contents(filtered_sections, request.businessIdea) # Create PDF with professional settings pdf = FPDF() pdf.set_auto_page_break(auto=True, margin=15) # Reduced margin to allow more charts per page pdf.add_page() # Professional color scheme - Grey, Black, and White primary_color = (0, 0, 0) # Black - for headings header_box_color = (128, 128, 128) # Darker gray - for front page header box accent_color = (0, 0, 0) # Black - for accent lines (changed from blue) text_color = (51, 51, 51) # Dark gray light_gray = (128, 128, 128) # Light gray # Professional fonts (fallback to Arial if not available) title_font = 'Arial' body_font = 'Arial' # Helper functions for consistent styling def add_header(pdf, text, size=24, y_offset=25, primary_color=(0,0,0), accent_color=(0,0,0)): pdf.set_font("Arial", "B", size) pdf.set_text_color(*primary_color) pdf.set_xy(25, y_offset) pdf.cell(0, 18, text, ln=True) # Calculate text width and make line autofit text_width = pdf.get_string_width(text) line_width = min(text_width + 10, 160) # Add 10mm padding, max 160mm (page width - margins) pdf.set_draw_color(200, 200, 200) # Light gray line pdf.set_line_width(0.3) # Thin line pdf.line(25, y_offset + 18, 25 + line_width, y_offset + 18) return y_offset + 25 def add_subheader(pdf, text, size=16, y_offset=None, primary_color=(0,0,0)): if y_offset is None: y_offset = pdf.get_y() + 10 pdf.set_font("Arial", "B", size) pdf.set_text_color(*primary_color) pdf.set_xy(25, y_offset) pdf.cell(0, 14, text, ln=True) # Calculate text width and make line autofit text_width = pdf.get_string_width(text) line_width = min(text_width + 8, 160) # Add 8mm padding, max 160mm (page width - margins) pdf.set_draw_color(200,200,200) pdf.set_line_width(0.4) pdf.line(25, y_offset + 14, 25 + line_width, y_offset + 14) return y_offset + 20 def add_body_text(pdf, text, y_offset=None, text_color=(51,51,51)): if y_offset is None: y_offset = pdf.get_y() + 5 pdf.set_font("Arial", "", 12) pdf.set_text_color(*text_color) pdf.set_xy(25, y_offset) pdf.multi_cell(160, 8, text) return pdf.get_y() def check_page_break(pdf, required_height=120, bottom_margin=300): """Check if we need a page break and return the Y position to use""" y = pdf.get_y() if y + required_height > bottom_margin: pdf.add_page() return 25 return y # ===== COVER PAGE ===== def add_cover_page(): # PURPLE BACKGROUND - Full page purple gradient effect pdf.set_fill_color(139, 92, 246) # Purple base pdf.rect(0, 0, 210, 297, 'F') # Full page purple # Add purple gradient effect with multiple rectangles pdf.set_fill_color(124, 58, 237) # Darker purple pdf.rect(0, 0, 210, 100, 'F') # Top section pdf.set_fill_color(147, 51, 234) # Medium purple pdf.rect(0, 100, 210, 100, 'F') # Middle section pdf.set_fill_color(168, 85, 247) # Lighter purple pdf.rect(0, 200, 210, 97, 'F') # Bottom section # Company/Project name - WHITE TEXT pdf.set_font(title_font, "B", 32) pdf.set_text_color(255, 255, 255) # Pure white business_name = request.businessIdea or "Business Plan" pdf.ln(15) pdf.cell(0, 25, business_name, ln=True, align='C') # Subtitle - WHITE TEXT pdf.set_font(title_font, "B", 18) pdf.set_text_color(255, 255, 255) # Pure white pdf.cell(0, 12, "Strategic Business Plan", ln=True, align='C') # Add some space pdf.ln(40) # Document info box - WHITE BACKGROUND with purple border pdf.set_fill_color(255, 255, 255) # White background pdf.set_draw_color(139, 92, 246) # Purple border pdf.set_line_width(2) # Thick purple border pdf.rect(25, 120, 160, 80, 'DF') # Filled with border pdf.set_font(body_font, "B", 14) pdf.set_text_color(139, 92, 246) # Purple text for title pdf.set_xy(35, 130) pdf.cell(0, 10, "Document Information", ln=True) pdf.set_font(body_font, "", 12) pdf.set_text_color(51, 51, 51) # Dark text for content pdf.set_xy(35, 145) pdf.cell(0, 8, f"Generated: {time.strftime('%B %d, %Y')}", ln=True) pdf.set_xy(35, 153) pdf.cell(0, 8, f"Format: {request.format.upper()}", ln=True) pdf.set_xy(35, 161) pdf.cell(0, 8, f"Sections: {len(request.sections)}", ln=True) pdf.set_xy(35, 169) pdf.cell(0, 8, f"Charts: {len(request.chartImages) if request.chartImages else 0}", ln=True) # Footer - WHITE TEXT on purple background pdf.set_font(body_font, "", 10) pdf.set_text_color(255, 255, 255) # White text pdf.set_xy(0, 280) pdf.cell(0, 8, "Confidential Business Document", ln=True, align='C') # ===== SECTION PAGES ===== def add_sections_with_charts(): # Parse charts from embedded chart data in section content embedded_charts = parse_embedded_chart_data(filtered_sections) if embedded_charts: # Create actual charts from the parsed data matplotlib_charts = [] chart_info_map = {} # Map section names to their charts for chart_info in embedded_charts: chart_bytes = create_chart_from_data(chart_info, request.businessIdea or "Business") if chart_bytes: matplotlib_charts.append(chart_bytes) # Map this chart to its source section source_section = chart_info["source_section"] if source_section not in chart_info_map: chart_info_map[source_section] = [] chart_info_map[source_section].append({ "chart_bytes": chart_bytes, "title": chart_info["title"], "type": chart_info["type"] }) else: pass else: matplotlib_charts = [] chart_info_map = {} total_charts_placed = 0 # Sort sections according to predefined order sorted_sections = sort_sections_by_order(filtered_sections) for i, section in enumerate(sorted_sections): # Always start each section on a new page pdf.add_page() # Accent line below header (not covering it) pdf.set_fill_color(*accent_color) pdf.rect(0, 15, 210, 2, 'F') # Below header, thinner line # Section title (properly positioned) pdf.set_font(title_font, "B", 20) pdf.set_text_color(*primary_color) pdf.set_xy(25, 25) # Back to normal position pdf.cell(0, 18, section.title, ln=True) # Add subtle line under section title - consistent with subheaders text_width = pdf.get_string_width(section.title) line_width = min(text_width + 8, 160) # Add 8mm padding, max 160mm pdf.set_draw_color(200, 200, 200) # Light gray line pdf.set_line_width(0.3) # Thin line pdf.line(25, pdf.get_y(), 25 + line_width, pdf.get_y()) # Consistent spacing after section title pdf.ln(8) # Section content (properly positioned) pdf.set_font(body_font, "", 12) pdf.set_text_color(*text_color) pdf.set_xy(25, pdf.get_y()) # Use current Y position for consistent spacing # Process content content_text = clean_text_for_pdf(section.content) content_text = convert_currency_symbols_to_iso(content_text) # Remove chart data markers content_text = re.sub(r'.*?', '', content_text, flags=re.DOTALL) content_text = re.sub(r'.*$', '', content_text, flags=re.DOTALL) # Remove the section title from content to prevent duplication (since we already added it above) section_title_clean = section.title.strip() content_lines = content_text.split('\n') filtered_lines = [] for line in content_lines: line_clean = line.strip() # Skip lines that match the section title (with or without markdown) if (line_clean == section_title_clean or line_clean == f"**{section_title_clean}**" or line_clean == f"# {section_title_clean}" or line_clean == f"## {section_title_clean}"): continue filtered_lines.append(line) content_text = '\n'.join(filtered_lines) # Enhance subheadings and remove duplicates content_text = enhance_subheadings_with_bold(content_text) content_text = remove_duplicate_headings(content_text) # Add content with proper markdown rendering if content_text and content_text.strip(): render_markdown_to_pdf(pdf, content_text, primary_color, accent_color, text_color) else: # Add placeholder text to prevent blank page pdf.multi_cell(160, 8, f"Content for {section.title} section is being processed...") # Add minimal spacer only if we have content if content_text and content_text.strip(): pdf.ln(3) # Small space after content # ---- Charts for this section ---- charts_added = 0 # IMPORTANT: initialize to avoid NameError section_charts = chart_info_map.get(section.title, []) if section_charts: # Group charts that can fit on the same page charts_per_page = 0 max_charts_per_page = 5 # Allow up to 5 charts per page for chart_info in section_charts: try: # Check if we need a new page for this chart chart_height_needed = 95 # Chart height (75) + title (20) + minimal spacing # Smart page break logic - allow multiple charts per page current_y = pdf.get_y() charts_per_page += 1 # Only force page break if absolutely necessary - allow up to 5 charts per page if (current_y + chart_height_needed > 290) or (charts_per_page > 5): pdf.add_page() # Header/footer removed for now y_title = 25 # Start at normal position charts_per_page = 1 # Reset counter for new page else: y_title = current_y # Add minimal spacing before chart if not at top of page if y_title > 25: # If not at top of page, add minimal space pdf.ln(3) # Minimal spacing to fit more charts per page # Subheader (title) and subtle rule y_title = add_subheader(pdf, chart_info["title"], 16, y_title) # Centered image with minimal border - optimized for more charts per page img_width = 100 # Optimized for smaller charts (6x4.5 ratio) x = (210 - img_width) / 2 y_img = y_title # Save image to temp file with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as tmp_img: tmp_img.write(chart_info["chart_bytes"]) temp_img_path = tmp_img.name # Chart dimensions (no border needed) chart_height = 75 # Optimized height for 6:4.5 aspect ratio (100 * 4.5/6) chart_width = img_width # Image pdf.image(temp_img_path, x=x, y=y_img, w=chart_width) # Move below image and clean up (spacing based on actual chart height) pdf.ln(chart_height + 5) # Chart height (75) + minimal spacing between charts try: os.remove(temp_img_path) except Exception: pass charts_added += 1 total_charts_placed += 1 except Exception as chart_error: continue else: pass # Each section is already on its own page, no need for complex page break logic # ===== BUILD THE PDF ===== try: # Add all sections try: add_cover_page() except Exception as e: raise Exception(f"Cover page failed: {str(e)}") try: add_table_of_contents_page(pdf, toc_items, request.businessIdea) except Exception as e: raise Exception(f"Table of contents failed: {str(e)}") # Executive summary is now handled as part of the sections # No separate page needed - the first section contains the executive summary try: add_sections_with_charts() except Exception as e: raise Exception(f"Sections with charts failed: {str(e)}") # CRITICAL CHECK: Ensure we have content in the PDF total_pages = pdf.page_no() if total_pages < 2: # Should have at least cover + 1 content page # Add a content page to prevent blank PDF pdf.add_page() pdf.set_font("Arial", "B", 16) pdf.set_text_color(255, 0, 0) pdf.cell(0, 20, "Content Processing Issue", ln=True, align='C') pdf.set_font("Arial", "", 12) pdf.set_text_color(0, 0, 0) pdf.cell(0, 10, "The PDF generation encountered an issue with content processing.", ln=True, align='C') pdf.cell(0, 10, "Please check the server logs for detailed information.", ln=True, align='C') # Get PDF output and convert to bytes try: pdf_output = pdf.output(dest='S') if isinstance(pdf_output, bytearray): pdf_bytes = bytes(pdf_output) else: pdf_bytes = pdf_output.encode('latin-1') if isinstance(pdf_output, str) else pdf_output # CRITICAL SAFETY CHECK: Ensure PDF is not empty if len(pdf_bytes) < 1000: # PDF should be at least 1KB # Try to add some content to prevent blank PDF pdf.add_page() pdf.set_font("Arial", "B", 16) pdf.set_text_color(255, 0, 0) # Red text pdf.cell(0, 20, "PDF Generation Issue Detected", ln=True, align='C') pdf.set_font("Arial", "", 12) pdf.set_text_color(0, 0, 0) pdf.cell(0, 10, "If you see this message, there was an issue with content processing.", ln=True, align='C') pdf.cell(0, 10, "Please check the server logs for details.", ln=True, align='C') # Regenerate PDF with error message pdf_output = pdf.output(dest='S') if isinstance(pdf_output, bytearray): pdf_bytes = bytes(pdf_output) else: pdf_bytes = pdf_output.encode('latin-1') if isinstance(pdf_output, str) else pdf_output return pdf_bytes, f"business_plan_{request.businessIdea or 'export'}.pdf" except Exception as e: raise Exception(f"PDF conversion failed: {str(e)}") except Exception as e: logging.error(f"PDF generation failed: {e}") raise Exception(f"Failed to generate professional PDF: {str(e)}") def create_word_document(request: ExportRequest): """Create a Word document from the business plan data with front page, TOC, and proper page breaks""" try: from docx import Document from docx.shared import Inches, Pt, RGBColor from docx.enum.text import WD_ALIGN_PARAGRAPH, WD_TAB_ALIGNMENT from docx.oxml import OxmlElement from docx.oxml.ns import qn except ImportError: raise Exception("python-docx is not installed. Please install it with: pip install python-docx") # Filter out empty sections to prevent blank pages filtered_sections = filter_empty_sections(request.sections) if not filtered_sections: raise Exception("No valid sections found after filtering. All sections appear to be empty.") doc = Document() # ===== FRONT PAGE ===== # Add top spacing doc.add_paragraph() doc.add_paragraph() # Business name as main title with enhanced styling - PURPLE THEME title = doc.add_heading(request.businessIdea or "Business Plan", 0) title.alignment = WD_ALIGN_PARAGRAPH.CENTER # Style the title with larger font and PURPLE color for run in title.runs: run.font.size = Pt(36) run.font.color.rgb = RGBColor(139, 92, 246) # PURPLE # Add decorative line under title - PURPLE doc.add_paragraph() line_para = doc.add_paragraph() line_para.paragraph_format.alignment = WD_ALIGN_PARAGRAPH.CENTER line_run = line_para.add_run("─" * 50) line_run.font.size = Pt(12) line_run.font.color.rgb = RGBColor(139, 92, 246) # PURPLE # Add subtitle with enhanced styling - PURPLE doc.add_paragraph() subtitle = doc.add_paragraph("Professional Business Plan Document") subtitle.alignment = WD_ALIGN_PARAGRAPH.CENTER subtitle_run = subtitle.runs[0] subtitle_run.font.size = Pt(18) subtitle_run.font.color.rgb = RGBColor(124, 58, 237) # DARKER PURPLE subtitle_run.italic = True # Add more spacing doc.add_paragraph() doc.add_paragraph() doc.add_paragraph() # Add document info in a styled box - PURPLE BORDER info_para = doc.add_paragraph() info_para.paragraph_format.alignment = WD_ALIGN_PARAGRAPH.CENTER # Add background color and border styling info_run = info_para.add_run("Prepared by: Business Plan Generator") info_run.font.size = Pt(14) info_run.font.color.rgb = RGBColor(139, 92, 246) # PURPLE doc.add_paragraph() date_para = doc.add_paragraph() date_para.paragraph_format.alignment = WD_ALIGN_PARAGRAPH.CENTER date_run = date_para.add_run(f"Date: {time.strftime('%B %Y')}") date_run.font.size = Pt(14) date_run.font.color.rgb = RGBColor(139, 92, 246) # PURPLE # Add more spacing before footer doc.add_paragraph() doc.add_paragraph() doc.add_paragraph() doc.add_paragraph() # Add footer with enhanced styling - PURPLE footer = doc.add_paragraph("Confidential Business Information") footer.alignment = WD_ALIGN_PARAGRAPH.CENTER footer_run = footer.runs[0] footer_run.font.size = Pt(12) footer_run.font.color.rgb = RGBColor(139, 92, 246) # PURPLE footer_run.bold = True # Add decorative line above footer - PURPLE doc.add_paragraph() footer_line_para = doc.add_paragraph() footer_line_para.paragraph_format.alignment = WD_ALIGN_PARAGRAPH.CENTER footer_line_run = footer_line_para.add_run("─" * 40) footer_line_run.font.size = Pt(8) footer_line_run.font.color.rgb = RGBColor(139, 92, 246) # PURPLE # ===== TABLE OF CONTENTS ===== # Add page break before TOC doc.add_page_break() # Generate TOC entries toc_items = generate_table_of_contents(filtered_sections, request.businessIdea or "Business") # Add TOC title toc_title = doc.add_heading("Table of Contents", level=1) toc_title.alignment = WD_ALIGN_PARAGRAPH.CENTER # Add decorative line under TOC title doc.add_paragraph() line_para = doc.add_paragraph() line_para.paragraph_format.alignment = WD_ALIGN_PARAGRAPH.CENTER line_run = line_para.add_run("─" * 40) # Decorative line line_run.font.size = Pt(8) line_run.font.color.rgb = None # Use default color # Add spacing before TOC entries doc.add_paragraph() # Add TOC entries with page numbers and dot leaders for item in toc_items: if item["level"] == 1: # Main section - bold, larger font p = doc.add_paragraph() p.paragraph_format.space_after = Pt(8) # Add spacing between main sections # Add section title title_run = p.add_run(item["title"]) title_run.bold = True title_run.font.size = Pt(12) # Add tab character and page number p.add_run("\t") page_run = p.add_run(f"{item['page']}") page_run.font.size = Pt(12) # Set tab position to right-align page numbers tab_stops = p.paragraph_format.tab_stops tab_stops.add_tab_stop(Inches(6.5), WD_TAB_ALIGNMENT.RIGHT) else: # Subsection - regular font, indented p = doc.add_paragraph() p.paragraph_format.space_after = Pt(4) # Smaller spacing for subsections p.paragraph_format.left_indent = Inches(0.3) # Indent subsections # Add subsection title subtitle_run = p.add_run(item["title"]) subtitle_run.font.size = Pt(10) # Add tab character and page number p.add_run("\t") page_run = p.add_run(f"{item['page']}") page_run.font.size = Pt(10) # Set tab position to right-align page numbers tab_stops = p.paragraph_format.tab_stops tab_stops.add_tab_stop(Inches(6.5), WD_TAB_ALIGNMENT.RIGHT) # Executive Summary is now handled as a regular section above # Sort sections according to predefined order sorted_sections = sort_sections_by_order(filtered_sections) # ===== MAIN SECTIONS ===== for i, section in enumerate(sorted_sections): if section.content.strip(): # Add page break before each section (including Executive Summary) doc.add_page_break() # Add section heading doc.add_heading(section.title, level=1) # Clean and process content clean_content = clean_text_for_pdf(section.content) clean_content = convert_currency_symbols_to_iso(clean_content) # Remove chart data markers clean_content = re.sub(r'.*?', '', clean_content, flags=re.DOTALL) clean_content = re.sub(r'.*$', '', clean_content, flags=re.DOTALL) # Remove the section title from content to prevent duplication section_title_clean = section.title.strip() content_lines = clean_content.split('\n') filtered_lines = [] for line in content_lines: line_clean = line.strip() # Skip lines that match the section title (with or without markdown) if (line_clean == section_title_clean or line_clean == f"**{section_title_clean}**" or line_clean == f"# {section_title_clean}" or line_clean == f"## {section_title_clean}"): continue filtered_lines.append(line) clean_content = '\n'.join(filtered_lines) # Render markdown content with proper heading styling render_markdown_to_docx(doc, clean_content) # Add charts for this section if they exist embedded_charts = parse_embedded_chart_data([section]) if embedded_charts: for chart_info in embedded_charts: try: chart_bytes = create_chart_from_data(chart_info, request.businessIdea or "Business") if chart_bytes: # Add chart title chart_title = chart_info["title"] doc.add_heading(f"{chart_title}", level=2) # Save image temporarily with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as tmp_img: tmp_img.write(chart_bytes) temp_img_path = tmp_img.name # Add image to Word document doc.add_picture(temp_img_path, width=Inches(6.0)) # 6 inches width # Clean up temp file os.remove(temp_img_path) except Exception as chart_error: logging.warning(f"Failed to add chart for section '{section.title}': {chart_error}") continue doc.add_paragraph() # Charts are now handled inline with the sections above # Save to bytes buffer = io.BytesIO() doc.save(buffer) buffer.seek(0) doc_bytes = buffer.getvalue() buffer.close() filename = f"business_plan_{request.businessIdea or 'export'}.docx" return doc_bytes, filename @app.post("/export", response_model=ExportResponse) async def export_business_plan(request: ExportRequest): """ Export a business plan in the specified format (PDF or Word). This endpoint creates documents and returns them as base64 data for frontend upload to UploadThing. FLOW: Server generates document → Returns base64 data → Frontend uploads to UploadThing → User gets download link """ try: # Generate unique request ID for tracking request_id = str(id(request)) # Add request fingerprint for duplicate detection request_fingerprint = f"{request.businessIdea}_{request.format}_{len(request.sections)}_{hash(str(request.sections))}" # Check if this is a duplicate request (within 5 seconds) current_time = time.time() if hasattr(export_business_plan, 'last_requests'): # Clean old requests (older than 5 seconds) export_business_plan.last_requests = { fp: timestamp for fp, timestamp in export_business_plan.last_requests.items() if current_time - timestamp < 5 } # Check for duplicate if request_fingerprint in export_business_plan.last_requests: raise HTTPException( status_code=429, detail="Duplicate export request detected. Please wait a few seconds before trying again." ) else: export_business_plan.last_requests = {} # Record this request export_business_plan.last_requests[request_fingerprint] = current_time # Validate request format if request.format not in ["pdf", "word"]: raise HTTPException(status_code=400, detail="Invalid format. Only 'pdf' and 'word' are supported.") # Validate required fields if not request.summary: raise HTTPException(status_code=400, detail="Summary is required for export.") if not request.sections: raise HTTPException(status_code=400, detail="Sections are required for export.") if len(request.sections) == 0: raise HTTPException(status_code=400, detail="At least one section is required for export.") # Check section content for i, section in enumerate(request.sections): if not section.content or not section.content.strip(): raise HTTPException(status_code=400, detail=f"Section '{section.title}' has no content.") logging.info(f"Exporting business plan for: {request.businessIdea} in {request.format.upper()} format") logging.info(f"Request has {len(request.sections)} sections and summary length: {len(request.summary)}") # Check required packages before proceeding try: if request.format == "pdf": import fpdf else: import docx except ImportError as import_error: raise HTTPException( status_code=500, detail=f"Required package not available: {str(import_error)}" ) # Create the document based on format try: if request.format == "pdf": doc_bytes, filename = create_pdf_document(request) else: # word doc_bytes, filename = create_word_document(request) except ImportError as import_error: raise HTTPException( status_code=500, detail=f"Missing required package for {request.format.upper()} generation: {str(import_error)}" ) except Exception as doc_error: logging.error(f"Document creation failed: {doc_error}") raise HTTPException( status_code=500, detail=f"Failed to create {request.format.upper()} document: {str(doc_error)}" ) # Return document data to frontend for UploadThing upload try: # Encode document bytes as base64 for frontend transmission import base64 document_data = base64.b64encode(doc_bytes).decode('utf-8') response = ExportResponse( success=True, downloadUrl="", # Will be set by frontend after UploadThing upload filename=filename, size=len(doc_bytes), message=f"Document generated successfully as {request.format.upper()}. Ready for frontend upload to UploadThing.", documentData=document_data # Base64 encoded document data ) return response except Exception as data_error: logging.error(f"Failed to prepare document data: {data_error}") raise HTTPException( status_code=500, detail=f"Failed to prepare document data: {str(data_error)}" ) except HTTPException: # Re-raise HTTP exceptions (validation errors) raise except Exception as e: # Catch any other unexpected errors error_message = f"Export failed: {str(e)}" logging.error(error_message) raise HTTPException(status_code=500, detail=error_message) @app.get("/download/{filename}") async def download_file(filename: str): """Download endpoint to serve the generated export files""" file_path = os.path.join("temp_exports", filename) if not os.path.exists(file_path): raise HTTPException(status_code=404, detail="File not found") return FileResponse(file_path, filename=filename) @app.get("/") def root(): html_content = """ Business Plan Generator

🚀 Business Plan Generator

AI-Powered Business Planning Made Simple

✅ FastAPI is running successfully
📊 Generate Business Plans
Create comprehensive business plans using AI models
💡 Get Suggestions
Receive AI-powered business suggestions and insights
📄 Export Documents
Export plans in PDF or Word format
💾 Database Integration
Persist and manage business plans with Supabase
PDF
Export Format
Word
Export Format
1.0.0
Version
HF Export
Features
""" from fastapi.responses import HTMLResponse return HTMLResponse(content=html_content) @app.get("/health") def health_check(): """ Health check endpoint to verify API connectivity and model availability. """ hf_token = os.getenv("HUGGINGFACEHUB_API_TOKEN") openai_api_key = os.getenv("OPENAI_API_KEY") status = { "api": "healthy", "models": {} } # Check Hugging Face models models_to_check = ["Mistral", "Qwen-2.5", "Llama", "Gemma", "Phi-3"] for model_name in models_to_check: try: # Just initialize the model without running inference if model_name == "": client = InferenceClient(provider="hf-inference", api_key=hf_token) # Just a simple check that the model exists client.model_info("mistralai/Mistral-7B-Instruct-v0.3") status["models"][model_name] = "available" elif model_name == "Qwen-2.5": # Check Qwen-2.5 availability client = InferenceClient(provider="hf-inference", api_key=hf_token) client.model_info("Qwen/Qwen2.5-7B-Instruct") status["models"][model_name] = "available" elif model_name == "Llama": # Check Llama availability client = InferenceClient(provider="hf-inference", api_key=hf_token) client.model_info("meta-llama/Llama-2-7b-chat-hf") status["models"][model_name] = "available" else: # Simplified check for other models status["models"][model_name] = "not checked" except Exception as e: status["models"][model_name] = f"error: {str(e)}" # Check OpenAI/GPT model if openai_api_key: try: status["models"]["GPT"] = "available" except Exception as e: status["models"]["GPT"] = f"error: {str(e)}" else: status["models"]["GPT"] = "unavailable (missing API key)" return status def fix_base64_padding(base64_string: str) -> str: """Fix common base64 padding issues""" # Remove any whitespace or newlines base64_string = base64_string.strip() # Add padding if needed missing_padding = len(base64_string) % 4 if missing_padding: base64_string += '=' * (4 - missing_padding) return base64_string def robust_base64_decode(base64_string: str) -> Optional[bytes]: """Robustly decode base64 string with error handling and padding correction""" try: # First try direct decoding return base64.b64decode(base64_string) except Exception as e1: try: # Try with padding correction fixed_string = fix_base64_padding(base64_string) return base64.b64decode(fixed_string) except Exception as e2: try: # Try with URL-safe base64 return base64.urlsafe_b64decode(fix_base64_padding(base64_string)) except Exception as e3: try: # Try with standard base64 ignoring padding return base64.b64decode(base64_string + '==', validate=False) except Exception as e4: return None def parse_embedded_chart_data(sections: List[ExportSection]) -> List[Dict[str, Any]]: """Parse chart data embedded in section content using CHART_DATA markers""" charts = [] for section in sections: if section.content: # Look for chart data markers chart_matches = re.findall(r'(.*?)', section.content, re.DOTALL) for chart_data in chart_matches: try: # Parse the chart data array chart_array = eval(chart_data.strip()) # Each chart is an array with: [Section, ChartType, Title, Period, Data] for chart_item in chart_array: if len(chart_item) >= 5: section_name = chart_item[0] chart_type = chart_item[1] # DC=Doughnut, BG=Bar Graph, etc. chart_title = chart_item[2] period = chart_item[3] data = chart_item[4] chart_info = { "section": section_name, "type": chart_type, "title": chart_title, "period": period, "data": data, "source_section": section.title } charts.append(chart_info) except Exception as e: continue return charts def create_chart_from_data(chart_info: Dict[str, Any], business_idea: str) -> bytes: """Create a matplotlib chart from parsed chart data""" try: chart_type = chart_info["type"] chart_title = chart_info["title"] data = chart_info["data"] if chart_type == "DC": # Doughnut Chart fig, ax = plt.subplots(figsize=(6, 4.5)) # Smaller size for better page fit # Extract labels and values from data labels = [item[0] for item in data] values = [item[1] for item in data] # Create doughnut chart wedges, texts, autotexts = ax.pie(values, labels=labels, autopct='%1.1f%%', startangle=90) ax.add_patch(plt.Circle((0, 0), 0.7, fc='white')) ax.set_title(f"{business_idea} - {chart_title}", fontsize=14, fontweight='bold', pad=15) # Smaller font elif chart_type == "BG": # Bar Graph fig, ax = plt.subplots(figsize=(6, 4.5)) # Smaller size for better page fit # Extract labels and values from data labels = [item[0] for item in data] values = [item[1] for item in data] # Create bar chart bars = ax.bar(labels, values, color=['#002060', '#FFD700', '#87CEEB', '#90EE90', '#FFB6C1']) ax.set_title(f"{business_idea} - {chart_title}", fontsize=14, fontweight='bold', pad=15) # Smaller font ax.set_ylabel('Value', fontsize=10) # Smaller font ax.set_xlabel('Category', fontsize=10) # Smaller font # Add value labels on bars for bar in bars: height = bar.get_height() ax.text(bar.get_x() + bar.get_width()/2., height, f'{height:,.0f}', ha='center', va='bottom', fontsize=9) # Smaller font plt.xticks(rotation=45) else: # Default to bar chart fig, ax = plt.subplots(figsize=(6, 4.5)) # Smaller size for better page fit labels = [item[0] for item in data] values = [item[1] for item in data] bars = ax.bar(labels, values) ax.set_title(f"{business_idea} - {chart_title}", fontsize=14, fontweight='bold', pad=15) # Smaller font plt.xticks(rotation=45) plt.tight_layout() # Save to bytes img_buffer = io.BytesIO() plt.savefig(img_buffer, format='png', dpi=300, bbox_inches='tight') img_buffer.seek(0) chart_bytes = img_buffer.getvalue() plt.close(fig) return chart_bytes except Exception as e: return None # Add these helper functions after the existing helper functions, before the create_pdf_document function def generate_table_of_contents(sections: List[ExportSection], business_idea: str) -> List[Dict[str, Any]]: """Generate a structured table of contents with page numbers following SECTION_ORDER""" toc_items = [] # Create a mapping from section titles to section objects for easy lookup section_map = {section.title: section for section in sections if section.content and section.content.strip()} # Add business plan sections in the predefined SECTION_ORDER for i, expected_title in enumerate(SECTION_ORDER): # Find the corresponding section section = section_map.get(expected_title) if section: # Each section starts on a new page, so page number is 3 + i (cover=1, TOC=2, sections start at 3) page_number = 3 + i toc_items.append({ "title": section.title, "level": 1, "page": page_number, "section_type": "section" }) # Extract subheadings from section content - limit to prevent TOC overflow subheadings = extract_subheadings(section.content) # Only include first 2 most important subheadings per section to keep TOC compact max_subheadings = min(2, len(subheadings)) # Reduced from 3 to 2 for j, subheading in enumerate(subheadings[:max_subheadings]): toc_items.append({ "title": subheading, "level": 2, "page": page_number, "section_type": "subsection" }) return toc_items def extract_subheadings(content: str) -> List[str]: """Extract subheadings from markdown content""" subheadings = [] lines = content.split('\n') for line in lines: line = line.strip() # Look for markdown headers (##, ###, etc.) if line.startswith('##') and not line.startswith('###'): # Remove markdown formatting and clean up subheading = re.sub(r'^##+\s*', '', line) subheading = re.sub(r'[*_`~]', '', subheading) if subheading.strip(): subheadings.append(subheading.strip()) return subheadings def add_table_of_contents_page(pdf, toc_items: List[Dict[str, Any]], business_idea: str): """Add a professional table of contents page following SECTION_ORDER""" # TOC items are already in the correct order from generate_table_of_contents sorted_toc_items = toc_items # Professional color scheme - Grey, Black, and White primary_color = (0, 0, 0) # Black - for headings header_box_color = (128, 128, 128) # Darker gray - for front page header box accent_color = (0, 0, 0) # Black - for accent lines (changed from blue) text_color = (51, 51, 51) # Dark gray light_gray = (128, 128, 128) # Light gray # Page title - more compact pdf.set_font("Arial", "B", 20) # Reduced from 24 to 20 pdf.set_text_color(*primary_color) pdf.set_xy(25, 20) # Reduced from 25 to 20 pdf.cell(0, 15, "Table of Contents", ln=True) # Reduced from 18 to 15 # Calculate text width and make line autofit text_width = pdf.get_string_width("Table of Contents") line_width = min(text_width + 12, 160) # Add 12mm padding, max 160mm # Add subtle line under title - consistent styling pdf.set_draw_color(200, 200, 200) # Light gray line pdf.set_line_width(0.3) # Thin line pdf.line(25, 35, 25 + line_width, 35) # TOC content - optimized for single page layout y_position = 50 # Reduced from 60 to 50 pages_used = 1 for item in sorted_toc_items: # Only add new page if absolutely necessary (very close to bottom) # Increased threshold significantly to force single-page TOC if y_position > 290: # Very close to bottom margin pdf.add_page() y_position = 25 pages_used += 1 # Indent based on level indent = 0 if item["level"] == 1 else 15 # Font and color based on level if item["level"] == 1: pdf.set_font("Arial", "B", 12) pdf.set_text_color(*primary_color) else: pdf.set_font("Arial", "", 10) pdf.set_text_color(*text_color) # Title pdf.set_xy(25 + indent, y_position) pdf.cell(120, 8, item["title"], ln=False) # Dots leading to page number pdf.set_font("Arial", "", 8) pdf.set_text_color(*light_gray) # Calculate dots position title_width = pdf.get_string_width(item["title"]) dots_start = 25 + indent + title_width + 5 dots_end = 160 # Draw dots for x in range(int(dots_start), int(dots_end), 3): pdf.set_xy(x, y_position + 3) pdf.cell(1, 1, ".", ln=False) # Page number pdf.set_xy(160, y_position) pdf.cell(0, 8, str(item["page"]), ln=False) # Reduced spacing between items for better page utilization y_position += 8 # Reduced from 10 to 8 for even tighter spacing # Add some spacing at the end pdf.ln(20) def filter_empty_sections(sections: List[ExportSection]) -> List[ExportSection]: """Filter out sections with no content to prevent empty pages""" filtered_sections = [] for section in sections: if section.content and section.content.strip(): # Clean the content to check if it's actually meaningful cleaned_content = clean_text_for_pdf(section.content) if cleaned_content and len(cleaned_content.strip()) > 10: # At least 10 characters filtered_sections.append(section) else: pass else: pass return filtered_sections def enhance_subheadings_with_bold(content: str) -> str: """Enhance subheadings with bold formatting and ensure no duplicates""" if not content: return content lines = content.split('\n') enhanced_lines = [] seen_headings = set() for line in lines: line = line.strip() if not line: enhanced_lines.append('') continue # Check if this is a subheading (## or ###) if line.startswith('##') and not line.startswith('###'): # Extract the heading text heading_text = re.sub(r'^##+\s*', '', line) heading_text = re.sub(r'[*_`~]', '', heading_text).strip() # Check for duplicates if heading_text.lower() in seen_headings: continue seen_headings.add(heading_text.lower()) # Make it bold and ensure proper formatting enhanced_line = f"**{heading_text}**" enhanced_lines.append(enhanced_line) else: enhanced_lines.append(line) return '\n'.join(enhanced_lines) def remove_duplicate_headings(content: str) -> str: """Remove duplicate headings from content""" if not content: return content lines = content.split('\n') cleaned_lines = [] seen_headings = set() for line in lines: line = line.strip() if not line: cleaned_lines.append('') continue # Check if this is any type of heading if line.startswith('#'): heading_text = re.sub(r'^#+\s*', '', line) heading_text = re.sub(r'[*_`~]', '', heading_text).strip() # Check for duplicates (case-insensitive) if heading_text.lower() in seen_headings: continue seen_headings.add(heading_text.lower()) cleaned_lines.append(line) else: cleaned_lines.append(line) return '\n'.join(cleaned_lines) def render_markdown_to_pdf(pdf, content: str, primary_color, accent_color, text_color): """Render markdown content to PDF with proper heading styling""" if not content: return lines = content.split('\n') current_y = pdf.get_y() for line in lines: line = line.strip() if not line: pdf.ln(3) # Consistent space for empty lines continue # Handle different heading levels if line.startswith("# "): # H1 - Main heading (like "Company Profile") text = line[2:].strip() pdf.set_font("Arial", "B", 20) # Bold, size 20 pdf.set_text_color(*primary_color) # Black pdf.set_xy(25, current_y) pdf.multi_cell(160, 10, text) # Calculate text width and make line autofit text_width = pdf.get_string_width(text) line_width = min(text_width + 12, 160) # Add 12mm padding, max 160mm # Add subtle line under main heading - consistent with section titles pdf.set_draw_color(200, 200, 200) # Light gray line pdf.set_line_width(0.3) # Thin line pdf.line(25, current_y + 10, 25 + line_width, current_y + 10) current_y = pdf.get_y() pdf.ln(8) # Consistent space after main heading elif line.startswith("## ") or re.match(r'^\d+\.\d+\s+', line): # H2 - Subheading or numbered subheading text = line[3:].strip() if line.startswith("## ") else line.strip() pdf.set_font("Arial", "B", 16) # Bold, size 16 pdf.set_text_color(*primary_color) # Black pdf.set_xy(25, current_y) pdf.multi_cell(160, 8, text) # Calculate text width and make line autofit text_width = pdf.get_string_width(text) line_width = min(text_width + 10, 160) # Add 10mm padding, max 160mm # Add subtle line under subheading - consistent styling pdf.set_draw_color(200, 200, 200) # Light gray line pdf.set_line_width(0.3) # Thin line pdf.line(25, current_y + 8, 25 + line_width, current_y + 8) current_y = pdf.get_y() pdf.ln(6) # Consistent space after subheading elif re.match(r'^\d+\.\d+\s+[A-Z]', line): # Numbered subheadings like "3.6 SWOT Analysis" text = line.strip() pdf.set_font("Arial", "B", 16) # Bold, size 16 pdf.set_text_color(*primary_color) # Black pdf.set_xy(25, current_y) pdf.multi_cell(160, 8, text) # Calculate text width and make line autofit text_width = pdf.get_string_width(text) line_width = min(text_width + 10, 160) # Add 10mm padding, max 160mm # Add subtle line under subheading - consistent styling pdf.set_draw_color(200, 200, 200) # Light gray line pdf.set_line_width(0.3) # Thin line pdf.line(25, current_y + 8, 25 + line_width, current_y + 8) current_y = pdf.get_y() pdf.ln(6) # Consistent space after subheading elif line.startswith("### ") or line.startswith("#### "): # H3/H4 - Smaller subheadings text = line[4:].strip() if line.startswith("### ") else line[5:].strip() pdf.set_font("Arial", "B", 14) # Bold, size 14 pdf.set_text_color(*primary_color) # Black pdf.set_xy(25, current_y) pdf.multi_cell(160, 8, text) # Calculate text width and make line autofit text_width = pdf.get_string_width(text) line_width = min(text_width + 8, 160) # Add 8mm padding, max 160mm # Subtle accent line - consistent styling pdf.set_draw_color(200, 200, 200) # Light gray line pdf.set_line_width(0.2) # Very thin line pdf.line(25, current_y + 8, 25 + line_width, current_y + 8) current_y = pdf.get_y() pdf.ln(4) # Consistent space after subheading elif line.startswith("**") and line.endswith("**"): # Bold text that might be a heading text = line[2:-2].strip() # Remove ** markers # Check if this looks like a main heading (no numbers, not too long) if not re.match(r'^\d+\.', text) and len(text) < 50: pdf.set_font("Arial", "B", 18) # Bold, size 18 pdf.set_text_color(*primary_color) # Black pdf.set_xy(25, current_y) pdf.multi_cell(160, 9, text) # Calculate text width and make line autofit text_width = pdf.get_string_width(text) line_width = min(text_width + 10, 160) # Add 10mm padding, max 160mm # Add subtle line under heading - consistent styling pdf.set_draw_color(200, 200, 200) # Light gray line pdf.set_line_width(0.3) # Thin line pdf.line(25, current_y + 2, 25 + line_width, current_y + 2) current_y = pdf.get_y() pdf.ln(6) # Consistent space after heading else: # Regular bold text pdf.set_font("Arial", "B", 12) # Bold, size 12 pdf.set_text_color(*text_color) # Dark gray pdf.set_xy(25, current_y) pdf.multi_cell(160, 8, text) current_y = pdf.get_y() else: # Normal paragraph text pdf.set_font("Arial", "", 12) # Regular font, size 12 pdf.set_text_color(*text_color) # Dark gray pdf.set_xy(25, current_y) pdf.multi_cell(160, 8, line) current_y = pdf.get_y() # Canonical currency map for symbol resolution (matches TypeScript version) canonical_currency_map = { # Standard currency symbols "$": "USD", "£": "GBP", "€": "EUR", "¥": "JPY", "₹": "INR", "₽": "RUB", "₩": "KRW", "₪": "ILS", "₺": "TRY", "₴": "UAH", "₼": "AZN", "₸": "KZT", "₾": "GEL", "֏": "AMD", "₲": "PYG", "₡": "CRC", "₣": "XPF", "₦": "NGN", "₵": "GHS", "₨": "PKR", "₱": "PHP", "₫": "VND", "₭": "LAK", "៛": "KHR", "৳": "BDT", "؋": "AFN", "﷼": "IRR", # Custom symbols "‡": "CUSTOM_DAGGER", "†": "CUSTOM_DAGGER_SINGLE", # Letter-based currencies "R": "ZAR", # South African Rand "C$": "CAD", # Canadian Dollar "A$": "AUD", # Australian Dollar "R$": "BRL", # Brazilian Real "S/": "PEN", # Peruvian Sol "Bs": "BOB", # Bolivian Boliviano "Q": "GTQ", # Guatemalan Quetzal "L": "HNL", # Honduran Lempira "ƒ": "AWG", # Aruban Florin "Vt": "VUV", # Vanuatu Vatu "T$": "TOP", # Tongan Pa'anga "T": "WST", # Samoan Tala "лв": "BGN", # Bulgarian Lev "Kč": "CZK", # Czech Koruna "kr": "DKK", # Danish Krone "Ft": "HUF", # Hungarian Forint "zł": "PLN", # Polish Zloty "lei": "RON", # Romanian Leu "CHF": "CHF", # Swiss Franc "KM": "BAM", # Bosnian Marka "ден": "MKD", # North Macedonian Denar "дин": "RSD", # Serbian Dinar "so'm": "UZS", # Uzbekistani Som "NT$": "TWD", # Taiwan Dollar "HK$": "HKD", # Hong Kong Dollar "S$": "SGD", # Singapore Dollar "RM": "MYR", # Malaysian Ringgit "฿": "THB", # Thai Baht "Rp": "IDR", # Indonesian Rupiah "K": "MMK", # Myanmar Kyat "Rs": "LKR", # Sri Lankan Rupee "Nu": "BTN", # Bhutanese Ngultrum "MVR": "MVR", # Maldivian Rufiyaa # Middle Eastern currencies "ع.د": "IQD", # Iraqi Dinar "ل.ل": "LBP", # Lebanese Pound "د.ا": "JOD", # Jordanian Dinar "ر.س": "SAR", # Saudi Riyal "د.إ": "AED", # UAE Dirham "ر.ق": "QAR", # Qatari Riyal "د.ك": "KWD", # Kuwaiti Dinar "د.ب": "BHD", # Bahraini Dinar "ر.ع.": "OMR", # Omani Rial "ر.ي": "YER", # Yemeni Rial "ج.م": "EGP", # Egyptian Pound "ل.د": "LYD", # Libyan Dinar "د.ت": "TND", # Tunisian Dinar "د.ج": "DZD", # Algerian Dinar "د.م.": "MAD", # Moroccan Dirham "ج.س.": "SDG", # Sudanese Pound "SSP": "SSP", # South Sudanese Pound # African currencies "KSh": "KES", # Kenyan Shilling "USh": "UGX", # Ugandan Shilling "TSh": "TZS", # Tanzanian Shilling "FRw": "RWF", # Rwandan Franc "FBu": "BIF", # Burundian Franc "MT": "MZN", # Mozambican Metical "P": "BWP", # Botswanan Pula "N$": "NAD", # Namibian Dollar "Ar": "MGA", # Malagasy Ariary "CF": "KMF", # Comorian Franc "MK": "MWK", # Malawian Kwacha "ZK": "ZMW", # Zambian Kwacha "Kz": "AOA", # Angolan Kwanza "FC": "CDF", # Congolese Franc "FCFA": "XAF", # Central African Franc "Db": "STD", # Sao Tomean Dobra "D": "GMD", # Gambian Dalasi "FG": "GNF", # Guinean Franc "CFA": "XOF", # West African Franc "Le": "SLL", # Sierra Leonean Leone "L$": "LRD", # Liberian Dollar "Br": "ETB", # Ethiopian Birr "Nfk": "ERN", # Eritrean Nakfa "Fdj": "DJF", # Djiboutian Franc "S": "SOS", # Somali Shilling } def convert_currency_symbols_to_iso(text: str) -> str: """ Convert currency symbols to ISO codes for PDF compatibility Matches the TypeScript canonicalCurrencyMap logic """ import re # Import at function level to avoid scope issues processed_text = text # Track all conversions for detailed logging conversions_made = [] total_conversions = 0 # Process each currency symbol in the canonical map for symbol, iso_code in canonical_currency_map.items(): if symbol in processed_text: # Check if it's a single character is_single_character = len(symbol) == 1 is_unicode_currency = is_single_character and ord(symbol) > 127 # Non-ASCII is_latin_single_letter = is_single_character and symbol.isalpha() and ord(symbol) < 128 # Handle multi-character symbols (like "Le", "KSh", etc.) is_multi_char_symbol = len(symbol) > 1 # Skip plain Latin single letters (to avoid converting words like "A", "R", etc.) # but DO convert single-character Unicode currency symbols like €, £, ¥, ؋, etc. if not is_unicode_currency and is_latin_single_letter: # Only convert if it's followed by a number (currency context) # Pattern: R500, R 500, R500,000 etc. pattern = rf'\b{re.escape(symbol)}\s*(\d{{1,3}}(?:,?\d{{3}})*(?:\.\d{{2}})?)\b' matches = re.findall(pattern, processed_text) if matches: old_text = processed_text processed_text = re.sub(pattern, rf'{iso_code} \1', processed_text) if old_text != processed_text: conversions_made.append(f"{symbol} → {iso_code} (context: {matches})") total_conversions += len(matches) continue # For all other symbols (Unicode currency symbols, multi-character symbols) # Count occurrences before replacement old_count = processed_text.count(symbol) # Replace all occurrences old_text = processed_text processed_text = processed_text.replace(symbol, iso_code) # Check if replacement actually happened if old_text != processed_text: conversions_made.append(f"{symbol} → {iso_code} ({old_count} instances)") total_conversions += old_count return processed_text def clean_unicode_for_pdf(text: str) -> str: # Replace problematic Unicode characters with ASCII equivalents unicode_replacements = { '\u2019': "'", # Right single quotation mark '\u2018': "'", # Left single quotation mark '\u201C': '"', # Left double quotation mark '\u201D': '"', # Right double quotation mark '\u2013': '-', # En dash '\u2014': '--', # Em dash '\u2022': '•', # Bullet '\u2026': '...', # Horizontal ellipsis '\u00A0': ' ', # Non-breaking space '\u00B0': '°', # Degree sign '\u00AE': '(R)', # Registered trademark '\u2122': '(TM)', # Trademark '\u00A9': '(C)', # Copyright } for unicode_char, replacement in unicode_replacements.items(): text = text.replace(unicode_char, replacement) # Simple text cleaning - keep basic punctuation and common symbols text = ''.join(char for char in text if ord(char) < 128 or char in '•°$€£¥₹₩₽₪₺₴₼₸₾֏₲₡₣₦₵₨₱₫₭៛৳؋﷼%') return text def render_markdown_to_docx(doc, content: str): """Render markdown content to Word document with proper heading styling""" if not content: return lines = content.split('\n') for line in lines: line = line.strip() if not line: doc.add_paragraph() # Empty paragraph for spacing continue # Handle different heading levels if line.startswith("# "): # H1 - Main heading (like "Company Profile") text = line[2:].strip() doc.add_heading(text, level=1) elif line.startswith("## ") or re.match(r'^\d+\.\d+\s+', line): # H2 - Subheading or numbered subheading text = line[3:].strip() if line.startswith("## ") else line.strip() doc.add_heading(text, level=2) elif re.match(r'^\d+\.\d+\s+[A-Z]', line): # Numbered subheadings like "3.6 SWOT Analysis" text = line.strip() doc.add_heading(text, level=2) elif line.startswith("### ") or line.startswith("#### "): # H3/H4 - Smaller subheadings text = line[4:].strip() if line.startswith("### ") else line[5:].strip() doc.add_heading(text, level=3) elif line.startswith("**") and line.endswith("**"): # Bold text that might be a heading text = line[2:-2].strip() # Remove ** markers # Check if this looks like a main heading (no numbers, not too long) if not re.match(r'^\d+\.', text) and len(text) < 50: doc.add_heading(text, level=1) # Treat as main heading else: # Regular bold text doc.add_paragraph(text) else: # Normal paragraph text doc.add_paragraph(line) def clean_text_for_docx(text: str) -> str: # Replace problematic Unicode characters with ASCII equivalents unicode_replacements = { '\u2019': "'", # Right single quotation mark '\u2018': "'", # Left single quotation mark '\u201C': '"', # Left double quotation mark '\u201D': '"', # Right double quotation mark '\u2013': '-', # En dash '\u2014': '--', # Em dash '\u2022': '•', # Bullet '\u2026': '...', # Horizontal ellipsis '\u00A0': ' ', # Non-breaking space '\u00B0': '°', # Degree sign '\u00AE': '(R)', # Registered trademark '\u2122': '(TM)', # Trademark '\u00A9': '(C)', # Copyright } for unicode_char, replacement in unicode_replacements.items(): text = text.replace(unicode_char, replacement) # Simple text cleaning - keep basic punctuation and common symbols text = ''.join(char for char in text if ord(char) < 128 or char in '•°$€£¥₹₩₽₪₺₴₼₸₾֏₲₡₣₦₵₨₱₫₭៛৳؋﷼%') return text