from fastapi import HTTPException from datetime import datetime from sqlalchemy.orm import Session from io import BytesIO import os from reportlab.lib.pagesizes import A4 from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, Image from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.units import inch from reportlab.lib import colors from app.dependency.authantication import JWTPayloadSchema from app.models.main.group import TblGroup from app.models.main.brand import TblBrand from app.models.main.segment import TblSegment from app.models.main.store_formate import TblStoreFormat from app.models.main.civil import TblCivil from app.models.main.display_racking_unit import TblDisplayRackingUnit from app.models.main.carpentry import TblCarpentry from app.models.main.electrical_cabling import TblElectricalCabling from app.models.main.display_board import TblDisplayBoard from app.models.main.commercial_equipment import TblCommercialEquipment from app.models.main.info_tech import TblInfoTech from app.models.main.visual_merchendising_elements import TblVisualMerchElements from app.models.main.Plumbing import TblPlumbing from app.models.main.additional_installations_fix import TblAdditionalInstallation from app.models.main.depreciation import TblDepreciation from app.models.main.summary import TblSummary, SummaryBase from app.api.module_1.schema import ( Brand, Segment, StoreFormat, Civil, DisplayRackingUnit, Carpentry, ElectricalCabling, DisplayBoard, CommercialEquipment, InfoTech, VisualMerchandisingElements, Plumbing, AdditionalInstallations, Depreciation, CapitalExpensesSummary, Module1ReportPreview, ) class Module1ReportService: def __init__(self, db: Session, token: JWTPayloadSchema): self.db = db self.token = token def _fetch_module1_data(self, group_id: int) -> Module1ReportPreview: # Validate group exists group = self.db.query(TblGroup).filter(TblGroup.group_id == group_id).first() if not group: raise HTTPException(status_code=404, detail=f"Group ID {group_id} not found in database") # Fetch all data for the group brand_data = self.db.query(TblBrand).filter_by(group_id=group_id).first() segment_data = self.db.query(TblSegment).filter_by(group_id=group_id).all() store_format_data = self.db.query(TblStoreFormat).filter_by(group_id=group_id).all() civil_data = self.db.query(TblCivil).filter_by(group_id=group_id).all() display_racking_data = self.db.query(TblDisplayRackingUnit).filter_by(group_id=group_id).all() carpentry_data = self.db.query(TblCarpentry).filter_by(group_id=group_id).all() electrical_cabling_data = self.db.query(TblElectricalCabling).filter_by(group_id=group_id).all() displayboard_data = self.db.query(TblDisplayBoard).filter_by(group_id=group_id).all() commercial_equipment_data = self.db.query(TblCommercialEquipment).filter_by(group_id=group_id).all() infotech_data = self.db.query(TblInfoTech).filter_by(group_id=group_id).all() visual_merch_data = self.db.query(TblVisualMerchElements).filter_by(group_id=group_id).all() plumbing_data = self.db.query(TblPlumbing).filter_by(group_id=group_id).all() additional_installation_data = self.db.query(TblAdditionalInstallation).filter_by(group_id=group_id).all() # depreciation_data = self.db.query(TblDepreciation).filter_by(group_id=group_id).first() depreciation_data = self.db.query(TblDepreciation).filter( TblDepreciation.group_id == group_id ).first() # Create depreciation object with proper field mapping ''' if depreciation_data: depreciation_obj = Depreciation( depreciation_id=depreciation_data.depreciation_id, group_id=depreciation_data.group_id, slm_years=depreciation_data.depreciation_years, depreciation_method=depreciation_data.depreciation_method, slm_remarks=depreciation_data.remark, other_methods=depreciation_data.other_methods if hasattr(depreciation_data, 'other_methods') else "N/A" ) else: depreciation_obj = Depreciation(group_id=group_id) ''' # Create depreciation object with proper field mapping print(f"DEBUG: Group {group_id} depreciation_data exists: {depreciation_data is not None}") if depreciation_data: print(f"DEBUG: Raw data - years: {depreciation_data.depreciation_years}, method: {depreciation_data.depreciation_method}") depreciation_obj = Depreciation( depreciation_id=depreciation_data.depreciation_id, group_id=depreciation_data.group_id, slm_years=depreciation_data.depreciation_years, depreciation_method=depreciation_data.depreciation_method, slm_remarks=depreciation_data.remark ) print(f"DEBUG: Created obj - years: {depreciation_obj.slm_years}, method: {depreciation_obj.depreciation_method}") else: depreciation_obj = Depreciation(group_id=group_id) print(f"DEBUG: No data found, using defaults") def sum_for_format(data_list, fmt_letter, attr="total"): return sum( (getattr(item, attr, 0) or 0) for item in (data_list or []) if fmt_letter in str(getattr(item, "store_format_type", "")) ) format_a_excl_it = ( sum_for_format(civil_data, "A") + sum_for_format(display_racking_data, "A") + sum_for_format(carpentry_data, "A") + sum_for_format(electrical_cabling_data, "A") + sum_for_format(displayboard_data, "A") + sum_for_format(commercial_equipment_data, "A") + sum_for_format(visual_merch_data, "A", "total_cost") + sum_for_format(plumbing_data, "A", "total_cost") + sum_for_format(additional_installation_data, "A", "total_cost") ) format_a_it = sum_for_format(infotech_data, "A") format_a_total = format_a_excl_it + format_a_it format_b_excl_it = ( sum_for_format(civil_data, "B") + sum_for_format(display_racking_data, "B") + sum_for_format(carpentry_data, "B") + sum_for_format(electrical_cabling_data, "B") + sum_for_format(displayboard_data, "B") + sum_for_format(commercial_equipment_data, "B") + sum_for_format(visual_merch_data, "B", "total_cost") + sum_for_format(plumbing_data, "B", "total_cost") + sum_for_format(additional_installation_data, "B", "total_cost") ) format_b_it = sum_for_format(infotech_data, "B") format_b_total = format_b_excl_it + format_b_it capital_expenses_summary = CapitalExpensesSummary( group_id=group_id, store_format_a_capital_excl_it=format_a_excl_it or 0.0, store_format_a_it_capital=format_a_it or 0.0, store_format_a_total=format_a_total or 0.0, store_format_b_capital_excl_it=format_b_excl_it or 0.0, store_format_b_it_capital=format_b_it or 0.0, store_format_b_total=format_b_total or 0.0, additional_remarks="Auto-generated dynamically", key_takeaways="CAPEX summary generated based on available data.", ) report_data = Module1ReportPreview( group_id=group_id, brand=Brand.from_orm(brand_data) if brand_data else Brand(), segments=[Segment.from_orm(s) for s in (segment_data or [])], store_formats=[StoreFormat.from_orm(sf) for sf in (store_format_data or [])], civil=[Civil.from_orm(c) for c in (civil_data or [])], display_racking_units=[DisplayRackingUnit.from_orm(dr) for dr in (display_racking_data or [])], carpentry=[Carpentry.from_orm(c) for c in (carpentry_data or [])], electrical_cabling=[ElectricalCabling.from_orm(ec) for ec in (electrical_cabling_data or [])], display_boards=[DisplayBoard.from_orm(db) for db in (displayboard_data or [])], commercial_equipment=[CommercialEquipment.from_orm(ce) for ce in (commercial_equipment_data or [])], infotech=[InfoTech.from_orm(it) for it in (infotech_data or [])], visual_merchandising=[VisualMerchandisingElements.from_orm(vm) for vm in (visual_merch_data or [])], plumbing=[Plumbing.from_orm(p) for p in (plumbing_data or [])], additional_installations=[AdditionalInstallations.from_orm(ai) for ai in (additional_installation_data or [])], depreciation=depreciation_obj, capital_expenses_summary=capital_expenses_summary, last_updated=datetime.now(), ) return report_data def _generate_pdf_report(self, data: Module1ReportPreview) -> str: """Generate comprehensive PDF report with all JSON response data""" buffer = BytesIO() doc = SimpleDocTemplate(buffer, pagesize=A4, leftMargin=30, rightMargin=30, topMargin=30, bottomMargin=30) styles = getSampleStyleSheet() story = [] # Logo try: logo = Image("TS Logo.png", width=150, height=75) logo.hAlign = 'LEFT' story.append(logo) story.append(Spacer(1, 10)) except: pass # Continue without logo if file not found # Title title_style = ParagraphStyle('CustomTitle', parent=styles['Heading1'], fontSize=14, spaceAfter=15, alignment=1, fontName='Helvetica-Bold') story.append(Paragraph("Module 1: Complete Report Data", title_style)) story.append(Spacer(1, 15)) section_style = ParagraphStyle('SectionHeader', parent=styles['Heading2'], fontSize=11, spaceAfter=5, spaceBefore=10, fontName='Helvetica-Bold') # Brand story.append(Paragraph("Brand Information", section_style)) brand_data = [['Field', 'Value'], ['Brand Name', data.brand.brand_name], ['Brand Image', data.brand.brand_image], ['Rationale', data.brand.rationale]] brand_table = Table(brand_data) brand_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(brand_table) story.append(Spacer(1, 10)) # Segments story.append(Paragraph("Segments", section_style)) for segment in data.segments: seg_data = [['Segment Name', segment.segment_name], ['Location', segment.location], ['Chief Wage Earner Occupation', segment.chief_wage_earner_occupation], ['Male Education', segment.male_education], ['Female Education', segment.female_education], ['ISEC Value', str(segment.isec_value)], ['Total Population', str(segment.total_population)], ['Additional Details', segment.additional_details]] seg_table = Table(seg_data) seg_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(seg_table) story.append(Spacer(1, 8)) # Store Formats story.append(Paragraph("Store Formats", section_style)) for store in data.store_formats: store_data = [['Store Name', store.store_name], ['Store Name Rationale', store.store_name_rationale], ['Store Size', str(store.store_size)], ['Merchandise', store.merchandise], ['Location', store.location], ['Additional Location Parameters', store.additional_location_parameters], ['Service Parameters', store.service_parameters], ['Technology Adoption', store.technology_adoption], ['Other Parameters', store.other_parameters], ['Store Format Type', store.store_format_type]] store_table = Table(store_data) store_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(store_table) story.append(Spacer(1, 8)) # Civil Work if data.civil: story.append(Paragraph("Civil Work", section_style)) civil_headers = ['Store Format Type', 'Type of Civil Work', 'Rate Per Sqft', 'Total Sqft', 'Total', 'Remarks'] civil_rows = [civil_headers] for civil in data.civil: civil_rows.append([civil.store_format_type, civil.type_of_civil_work, str(civil.rate_per_sqft), str(civil.total_sqft), str(civil.total), civil.remarks]) civil_table = Table(civil_rows) civil_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 7), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(civil_table) story.append(Spacer(1, 10)) # Display Racking Units if data.display_racking_units: story.append(Paragraph("Display Racking Units", section_style)) rack_headers = ['Store Format Type', 'Type Display Racking Unit', 'Number of Units', 'Cost Per Unit', 'Total', 'Remarks'] rack_rows = [rack_headers] for rack in data.display_racking_units: rack_rows.append([rack.store_format_type, rack.type_display_racking_unit, str(rack.number_of_units), str(rack.cost_per_unit), str(rack.total), rack.remarks]) rack_table = Table(rack_rows) rack_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 7), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(rack_table) story.append(Spacer(1, 10)) # Carpentry if data.carpentry: story.append(Paragraph("Carpentry", section_style)) carp_headers = ['Store Format Type', 'Type', 'Units', 'Cost Per Unit', 'Total', 'Remarks'] carp_rows = [carp_headers] for carp in data.carpentry: carp_rows.append([carp.store_format_type, carp.type, str(carp.units), str(carp.cost_per_unit), str(carp.total), carp.remarks]) carp_table = Table(carp_rows) carp_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True), ('LEFTPADDING', (0, 0), (-1, -1), 6), ('RIGHTPADDING', (0, 0), (-1, -1), 6), ('TOPPADDING', (0, 0), (-1, -1), 6), ('BOTTOMPADDING', (0, 0), (-1, -1), 6) ])) story.append(carp_table) story.append(Spacer(1, 10)) # Electrical Cabling if data.electrical_cabling: story.append(Paragraph("Electrical Cabling", section_style)) elec_headers = ['Store Format Type', 'Feature', 'Details', 'Rate Per Sqft/Unit', 'Total', 'Remarks'] elec_rows = [elec_headers] for elec in data.electrical_cabling: elec_rows.append([elec.store_format_type, elec.feature, elec.details, str(elec.rate_per_sqft_or_unit), str(elec.total), elec.remarks]) elec_table = Table(elec_rows) elec_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True), ('LEFTPADDING', (0, 0), (-1, -1), 6), ('RIGHTPADDING', (0, 0), (-1, -1), 6), ('TOPPADDING', (0, 0), (-1, -1), 6), ('BOTTOMPADDING', (0, 0), (-1, -1), 6) ])) story.append(elec_table) story.append(Spacer(1, 10)) # Display Boards if data.display_boards: story.append(Paragraph("Display Boards", section_style)) disp_headers = ['Store Format Type', 'Type', 'Size', 'Pricing Type', 'Cost Rate', 'Units', 'Total', 'Remarks'] disp_rows = [disp_headers] for disp in data.display_boards: disp_rows.append([disp.store_format_type, disp.type, str(disp.size), disp.pricing_type, str(disp.cost_rate), str(disp.units), str(disp.total), disp.remarks]) disp_table = Table(disp_rows) disp_table.setStyle(TableStyle([ ('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True), ('LEFTPADDING', (0, 0), (-1, -1), 6), ('RIGHTPADDING', (0, 0), (-1, -1), 6), ('TOPPADDING', (0, 0), (-1, -1), 6), ('BOTTOMPADDING', (0, 0), (-1, -1), 6) ])) story.append(disp_table) story.append(Spacer(1, 10)) # Commercial Equipment if data.commercial_equipment: story.append(Paragraph("Commercial Equipment", section_style)) comm_headers = ['Store Format Type', 'Type of Equipment', 'Units', 'Cost Per Unit', 'Equipment Type', 'Total', 'Remarks'] comm_rows = [comm_headers] for comm in data.commercial_equipment: comm_rows.append([comm.store_format_type, comm.type_of_equipment, str(comm.units), str(comm.cost_per_unit), comm.equipment_type, str(comm.total), comm.remarks]) comm_table = Table(comm_rows) comm_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 6), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(comm_table) story.append(Spacer(1, 10)) # InfoTech if data.infotech: story.append(Paragraph("InfoTech", section_style)) it_headers = ['Store Format Type', 'Type', 'Category', 'Units', 'Cost Per Unit', 'Total', 'Available Capital'] it_rows = [it_headers] for it in data.infotech: it_rows.append([it.store_format_type, it.type, it.category, str(it.units), str(it.cost_per_unit), str(it.total), str(it.available_capital)]) it_table = Table(it_rows) it_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 6), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(it_table) story.append(Spacer(1, 10)) # Visual Merchandising if data.visual_merchandising: story.append(Paragraph("Visual Merchandising", section_style)) vm_headers = ['Type', 'Units', 'Cost', 'Total', 'Internal Display', 'External Display', 'Frequency Change/Year'] vm_rows = [vm_headers] for vm in data.visual_merchandising: vm_rows.append([vm.type, str(vm.units), str(vm.cost), str(vm.total), vm.internal_display, vm.external_display, str(vm.frequency_change_per_year)]) vm_table = Table(vm_rows) vm_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 6), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(vm_table) story.append(Spacer(1, 10)) # Plumbing if data.plumbing: story.append(Paragraph("Plumbing", section_style)) plumb_headers = ['Type', 'Units', 'Cost Per Unit', 'Total', 'Remarks'] plumb_rows = [plumb_headers] for plumb in data.plumbing: plumb_rows.append([plumb.type, str(plumb.units), str(plumb.cost_per_unit), str(plumb.total), plumb.remarks]) plumb_table = Table(plumb_rows) plumb_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 7), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(plumb_table) story.append(Spacer(1, 10)) # Additional Installations if data.additional_installations: story.append(Paragraph("Additional Installations", section_style)) add_headers = ['Name', 'Description', 'Cost Per Unit', 'Total Cost', 'Remarks'] add_rows = [add_headers] for add in data.additional_installations: add_rows.append([add.name, add.description, str(add.cost_per_unit), str(add.total_cost), add.remarks]) add_table = Table(add_rows) add_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 7), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(add_table) story.append(Spacer(1, 10)) # Capital Expenses Summary story.append(Paragraph("Capital Expenses Summary", section_style)) capex_data = [['Field', 'Value'], ['Store Format A Capital Excl IT', f"Rs. {data.capital_expenses_summary.store_format_a_capital_excl_it:,.2f}"], ['Store Format A IT Capital', f"Rs. {data.capital_expenses_summary.store_format_a_it_capital:,.2f}"], ['Store Format A Total', f"Rs. {data.capital_expenses_summary.store_format_a_total:,.2f}"], ['Store Format B Capital Excl IT', f"Rs. {data.capital_expenses_summary.store_format_b_capital_excl_it:,.2f}"], ['Store Format B IT Capital', f"Rs. {data.capital_expenses_summary.store_format_b_it_capital:,.2f}"], ['Store Format B Total', f"Rs. {data.capital_expenses_summary.store_format_b_total:,.2f}"], ['Additional Remarks', data.capital_expenses_summary.additional_remarks], ['Key Takeaways', data.capital_expenses_summary.key_takeaways]] capex_table = Table(capex_data) capex_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(capex_table) # Depreciation if data.depreciation: story.append(Spacer(1, 10)) story.append(Paragraph("Depreciation Information", section_style)) dep_data = [['Field', 'Value'], ['SLM Years', str(data.depreciation.slm_years)], ['SLM Remarks', data.depreciation.slm_remarks], ['Other Methods', data.depreciation.other_methods], ['Depreciation Method', data.depreciation.depreciation_method]] dep_table = Table(dep_data) dep_table.setStyle(TableStyle([('ALIGN', (0, 0), (-1, -1), 'LEFT'), ('FONTNAME', (0, 0), (0, -1), 'Helvetica-Bold'), ('FONTSIZE', (0, 0), (-1, -1), 8), ('GRID', (0, 0), (-1, -1), 0.5, colors.black), ('VALIGN', (0, 0), (-1, -1), 'TOP'), ('WORDWRAP', (0, 0), (-1, -1), True)])) story.append(dep_table) # Last Updated story.append(Spacer(1, 10)) story.append(Paragraph(f"Last Updated: {data.last_updated.strftime('%Y-%m-%d %H:%M:%S')}", styles['Normal'])) doc.build(story) buffer.seek(0) # Save to specified directory timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"module1_complete_report_group_{data.group_id}_{timestamp}.pdf" filepath = f"uploaded_files/{filename}" os.makedirs(os.path.dirname(filepath), exist_ok=True) with open(filepath, 'wb') as f: f.write(buffer.getvalue()) return filepath def generate_and_save_pdf(self, group_id: int) -> str: """Generate PDF and save to database, return file path""" report_data = self._fetch_module1_data(group_id) file_path = self._generate_pdf_report(report_data) self._save_file_path_to_db(group_id, file_path) return file_path def _save_file_path_to_db(self, group_id: int, file_path: str): """Save or update file path in summary table with group_id_mod""" try: group_id_mod = f"g{group_id}_m1" print(f"DEBUG: Looking for existing record with group_id_mod: {group_id_mod}") existing_summary = self.db.query(TblSummary).filter(TblSummary.group_id_mod == group_id_mod).first() if existing_summary: print(f"DEBUG: Found existing record, updating uploaded_files") existing_summary.uploaded_files = file_path self.db.commit() self.db.refresh(existing_summary) print(f"DEBUG: Updated record successfully") else: print(f"DEBUG: No existing record found, creating new one") new_summary = TblSummary( group_id=group_id, group_id_mod=group_id_mod, uploaded_files=file_path ) self.db.add(new_summary) self.db.commit() self.db.refresh(new_summary) print(f"DEBUG: Created new record with summary_id: {new_summary.summary_id}") except Exception as e: self.db.rollback() print(f"ERROR saving file path to database: {e}") import traceback traceback.print_exc()