| import io | |
| import os | |
| import tempfile | |
| import pandas as pd | |
| import plotly.express as px | |
| import plotly.io as pio | |
| from datetime import datetime | |
| from typing import Dict, Optional | |
| from reportlab.lib import colors | |
| from reportlab.lib.pagesizes import A4 | |
| from reportlab.lib.units import inch | |
| from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer, Image, PageBreak | |
| from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle | |
| from config import get_chart_theme, DESIGN_SYSTEM | |
| from ai_engine import generate_ai_summary | |
| PDF_TRANSLATIONS = { | |
| 'English': { | |
| 'title': 'Production Monitor Dashboard', | |
| 'subtitle': 'Comprehensive Production Analysis Report', | |
| 'company': 'Nilsen Service & Consulting AS', | |
| 'division': 'Production Analytics Division', | |
| 'report_period': 'Report Period', | |
| 'generated': 'Generated', | |
| 'total_records': 'Total Records', | |
| 'executive_summary': 'Executive Summary', | |
| 'production_summary': 'Production Summary', | |
| 'production_analysis_charts': 'Production Analysis Charts', | |
| 'quality_control': 'Quality Control Analysis', | |
| 'intelligent_analysis': 'Intelligent Analysis', | |
| 'material_type': 'Material Type', | |
| 'total_kg': 'Total (kg)', | |
| 'share': 'Share (%)', | |
| 'daily_avg_kg': 'Daily Avg (kg)', | |
| 'material': 'Material', | |
| 'outliers': 'Outliers', | |
| 'normal_range_kg': 'Normal Range (kg)', | |
| 'status': 'Status', | |
| 'footer_generated': 'This report was generated by Production Monitor System', | |
| 'footer_records': 'Report contains {records} data records across {days} working days', | |
| 'summary_text': 'This report analyzes production data spanning <b>{days} working days</b>. Total output achieved: <b>{total} kg</b> with an average daily production of <b>{avg} kg</b>.', | |
| 'key_highlights': 'Key Highlights:', | |
| 'total_production': 'Total production', | |
| 'daily_average': 'Daily average', | |
| 'materials_tracked': 'Materials tracked', | |
| 'data_quality': 'Data quality', | |
| 'records_processed': 'records processed', | |
| 'production_distribution': 'Production Distribution by Material', | |
| 'daily_production_trend': 'Daily Production Trend', | |
| 'production_by_material': 'Production by Material Type', | |
| 'production_by_shift': 'Production by Shift', | |
| 'chart_analysis': 'Analysis', | |
| 'dist_insight': 'Material distribution shows production allocation across different materials.', | |
| 'trend_insight': 'Production trend reveals operational patterns and seasonal variations.', | |
| 'material_insight': 'Material comparison highlights performance differences and production capacities.', | |
| 'shift_insight': 'Shift analysis reveals operational efficiency differences between day and night operations.', | |
| 'charts_failed': 'Charts Generation Failed', | |
| 'data_summary': 'Production Data Summary:', | |
| 'status_good': 'GOOD', | |
| 'status_monitor': 'MONITOR', | |
| 'status_attention': 'ATTENTION' | |
| }, | |
| 'Norsk': { | |
| 'title': 'Produksjonsovervakingsdashboard', | |
| 'subtitle': 'Omfattende produksjonsanalyserapport', | |
| 'company': 'Nilsen Service & Consulting AS', | |
| 'division': 'Produksjonsanalyseavdeling', | |
| 'report_period': 'Rapportperiode', | |
| 'generated': 'Generert', | |
| 'total_records': 'Totalt antall poster', | |
| 'executive_summary': 'Oppsummering', | |
| 'production_summary': 'Produksjonsoversikt', | |
| 'production_analysis_charts': 'Produksjonsanalysediagrammer', | |
| 'quality_control': 'Kvalitetskontrollanalyse', | |
| 'intelligent_analysis': 'Intelligent analyse', | |
| 'material_type': 'Materialtype', | |
| 'total_kg': 'Totalt (kg)', | |
| 'share': 'Andel (%)', | |
| 'daily_avg_kg': 'Daglig gj.snitt (kg)', | |
| 'material': 'Material', | |
| 'outliers': 'Avvik', | |
| 'normal_range_kg': 'Normalomrade (kg)', | |
| 'status': 'Status', | |
| 'footer_generated': 'Denne rapporten ble generert av produksjonsovervakingssystemet', | |
| 'footer_records': 'Rapporten inneholder {records} dataposter over {days} arbeidsdager', | |
| 'summary_text': 'Denne rapporten analyserer produksjonsdata over <b>{days} arbeidsdager</b>. Total produksjon oppnadd: <b>{total} kg</b> med et daglig gjennomsnitt pa <b>{avg} kg</b>.', | |
| 'key_highlights': 'Hovedpunkter:', | |
| 'total_production': 'Total produksjon', | |
| 'daily_average': 'Daglig gjennomsnitt', | |
| 'materials_tracked': 'Materialer sporet', | |
| 'data_quality': 'Datakvalitet', | |
| 'records_processed': 'poster behandlet', | |
| 'production_distribution': 'Produksjonsfordeling etter material', | |
| 'daily_production_trend': 'Daglig produksjonstrend', | |
| 'production_by_material': 'Produksjon etter materialtype', | |
| 'production_by_shift': 'Produksjon etter skift', | |
| 'chart_analysis': 'Analyse', | |
| 'dist_insight': 'Materialfordeling viser produksjonsallokering pa tvers av ulike materialer.', | |
| 'trend_insight': 'Produksjonstrend avslorer operasjonelle monster og sesongvariasjoner.', | |
| 'material_insight': 'Materialsammenligning fremhever ytelsesforskjeller og produksjonskapasitet.', | |
| 'shift_insight': 'Skiftanalyse avslorer operasjonelle effektivitetsforskjeller mellom dag- og nattoperasjoner.', | |
| 'charts_failed': 'Diagramgenerering mislyktes', | |
| 'data_summary': 'Produksjonsdataoversikt:', | |
| 'status_good': 'GOD', | |
| 'status_monitor': 'OVERVAKNING', | |
| 'status_attention': 'OPPMERKSOMHET' | |
| } | |
| } | |
| def get_pdf_translation(lang: str = 'English') -> dict: | |
| return PDF_TRANSLATIONS.get(lang, PDF_TRANSLATIONS['English']) | |
| def save_chart_as_image(fig, filename: str) -> Optional[str]: | |
| try: | |
| temp_dir = tempfile.gettempdir() | |
| filepath = os.path.join(temp_dir, filename) | |
| theme = get_chart_theme()['layout'].copy() | |
| theme.update({ | |
| 'font': dict(size=12, family="Arial"), | |
| 'plot_bgcolor': 'white', | |
| 'paper_bgcolor': 'white', | |
| 'margin': dict(t=50, b=40, l=40, r=40) | |
| }) | |
| fig.update_layout(**theme) | |
| try: | |
| pio.write_image(fig, filepath, format='png', width=800, height=400, scale=2, engine='kaleido') | |
| if os.path.exists(filepath): | |
| return filepath | |
| except: | |
| pass | |
| return None | |
| except: | |
| return None | |
| def create_pdf_charts(df: pd.DataFrame, stats: Dict, t: dict) -> Dict[str, Optional[str]]: | |
| charts = {} | |
| materials = [k for k in stats.keys() if k != '_total_'] | |
| values = [stats[mat]['total'] for mat in materials] | |
| labels = [mat.replace('_', ' ').title() for mat in materials] | |
| if len(materials) > 0 and len(values) > 0: | |
| try: | |
| fig_pie = px.pie(values=values, names=labels, title=t['production_distribution']) | |
| charts['pie'] = save_chart_as_image(fig_pie, "distribution.png") | |
| except: | |
| pass | |
| if len(df) > 0: | |
| try: | |
| daily_data = df.groupby('date')['weight_kg'].sum().reset_index() | |
| if len(daily_data) > 0: | |
| fig_trend = px.line( | |
| daily_data, x='date', y='weight_kg', | |
| title=t['daily_production_trend'], | |
| labels={'date': 'Date', 'weight_kg': 'Weight (kg)'}, | |
| color_discrete_sequence=[DESIGN_SYSTEM['colors']['primary']] | |
| ) | |
| charts['trend'] = save_chart_as_image(fig_trend, "trend.png") | |
| except: | |
| pass | |
| if len(materials) > 0 and len(values) > 0: | |
| try: | |
| fig_bar = px.bar( | |
| x=labels, y=values, | |
| title=t['production_by_material'], | |
| labels={'x': t['material_type'], 'y': 'Weight (kg)'}, | |
| color_discrete_sequence=[DESIGN_SYSTEM['colors']['primary']] | |
| ) | |
| charts['bar'] = save_chart_as_image(fig_bar, "materials.png") | |
| except: | |
| pass | |
| if 'shift' in df.columns and len(df) > 0: | |
| try: | |
| shift_data = df.groupby('shift')['weight_kg'].sum().reset_index() | |
| if len(shift_data) > 0 and shift_data['weight_kg'].sum() > 0: | |
| fig_shift = px.pie(shift_data, values='weight_kg', names='shift', title=t['production_by_shift']) | |
| charts['shift'] = save_chart_as_image(fig_shift, "shifts.png") | |
| except: | |
| pass | |
| return charts | |
| def create_enhanced_pdf_report(df: pd.DataFrame, stats: Dict, outliers: Dict, model=None, lang: str = 'English') -> io.BytesIO: | |
| buffer = io.BytesIO() | |
| doc = SimpleDocTemplate(buffer, pagesize=A4, rightMargin=50, leftMargin=50, topMargin=50, bottomMargin=50) | |
| elements = [] | |
| styles = getSampleStyleSheet() | |
| t = get_pdf_translation(lang) | |
| title_style = ParagraphStyle( | |
| 'CustomTitle', | |
| parent=styles['Heading1'], | |
| fontSize=24, | |
| spaceAfter=30, | |
| alignment=1, | |
| textColor=colors.darkblue | |
| ) | |
| subtitle_style = ParagraphStyle( | |
| 'CustomSubtitle', | |
| parent=styles['Heading2'], | |
| fontSize=16, | |
| spaceAfter=20, | |
| textColor=colors.darkblue | |
| ) | |
| analysis_style = ParagraphStyle( | |
| 'AnalysisStyle', | |
| parent=styles['Normal'], | |
| fontSize=11, | |
| spaceAfter=12, | |
| leftIndent=20, | |
| textColor=colors.darkgreen | |
| ) | |
| elements.append(Spacer(1, 100)) | |
| elements.append(Paragraph(t['title'], title_style)) | |
| elements.append(Paragraph(t['subtitle'], styles['Heading3'])) | |
| elements.append(Spacer(1, 50)) | |
| report_info = f""" | |
| <para alignment="center"> | |
| <b>{t['company']}</b><br/> | |
| {t['division']}<br/><br/> | |
| <b>{t['report_period']}:</b> {df['date'].min().strftime('%B %d, %Y')} - {df['date'].max().strftime('%B %d, %Y')}<br/> | |
| <b>{t['generated']}:</b> {datetime.now().strftime('%B %d, %Y at %H:%M')}<br/> | |
| <b>{t['total_records']}:</b> {len(df):,} | |
| </para> | |
| """ | |
| elements.append(Paragraph(report_info, styles['Normal'])) | |
| elements.append(PageBreak()) | |
| elements.append(Paragraph(t['executive_summary'], subtitle_style)) | |
| total_production = stats['_total_']['total'] | |
| work_days = stats['_total_']['work_days'] | |
| daily_avg = stats['_total_']['daily_avg'] | |
| exec_summary = f""" | |
| <para> | |
| {t['summary_text'].format(days=work_days, total=f"{total_production:,.0f}", avg=f"{daily_avg:,.0f}")} | |
| <br/><br/> | |
| <b>{t['key_highlights']}</b><br/> | |
| • {t['total_production']}: {total_production:,.0f} kg<br/> | |
| • {t['daily_average']}: {daily_avg:,.0f} kg<br/> | |
| • {t['materials_tracked']}: {len([k for k in stats.keys() if k != '_total_'])}<br/> | |
| • {t['data_quality']}: {len(df):,} {t['records_processed']} | |
| </para> | |
| """ | |
| elements.append(Paragraph(exec_summary, styles['Normal'])) | |
| elements.append(Spacer(1, 20)) | |
| elements.append(Paragraph(t['production_summary'], styles['Heading3'])) | |
| summary_data = [[t['material_type'], t['total_kg'], t['share'], t['daily_avg_kg']]] | |
| for material, info in stats.items(): | |
| if material != '_total_': | |
| summary_data.append([ | |
| material.replace('_', ' ').title(), | |
| f"{info['total']:,.0f}", | |
| f"{info['percentage']:.1f}%", | |
| f"{info['daily_avg']:,.0f}" | |
| ]) | |
| summary_table = Table(summary_data, colWidths=[2*inch, 1.5*inch, 1*inch, 1.5*inch]) | |
| summary_table.setStyle(TableStyle([ | |
| ('BACKGROUND', (0, 0), (-1, 0), colors.darkblue), | |
| ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), | |
| ('ALIGN', (0, 0), (-1, -1), 'CENTER'), | |
| ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), | |
| ('GRID', (0, 0), (-1, -1), 1, colors.black), | |
| ('ROWBACKGROUNDS', (0, 1), (-1, -1), [colors.white, colors.lightgrey]) | |
| ])) | |
| elements.append(summary_table) | |
| elements.append(PageBreak()) | |
| elements.append(Paragraph(t['production_analysis_charts'], subtitle_style)) | |
| try: | |
| charts = create_pdf_charts(df, stats, t) | |
| except: | |
| charts = {} | |
| chart_insights = { | |
| 'pie': t['dist_insight'], | |
| 'trend': t['trend_insight'], | |
| 'bar': t['material_insight'], | |
| 'shift': t['shift_insight'] | |
| } | |
| chart_titles = { | |
| 'pie': t['production_distribution'], | |
| 'trend': t['daily_production_trend'], | |
| 'bar': t['production_by_material'], | |
| 'shift': t['production_by_shift'] | |
| } | |
| charts_added = False | |
| for chart_type in ['pie', 'trend', 'bar', 'shift']: | |
| chart_path = charts.get(chart_type) | |
| if chart_path and os.path.exists(chart_path): | |
| try: | |
| elements.append(Paragraph(chart_titles[chart_type], styles['Heading3'])) | |
| elements.append(Image(chart_path, width=6*inch, height=3*inch)) | |
| insight_text = f"<i>{t['chart_analysis']}: {chart_insights[chart_type]}</i>" | |
| elements.append(Paragraph(insight_text, analysis_style)) | |
| elements.append(Spacer(1, 20)) | |
| charts_added = True | |
| except: | |
| pass | |
| if not charts_added: | |
| elements.append(Paragraph(t['charts_failed'], styles['Heading3'])) | |
| elements.append(Paragraph(t['data_summary'], styles['Normal'])) | |
| for material, info in stats.items(): | |
| if material != '_total_': | |
| summary_text = f"• {material.replace('_', ' ').title()}: {info['total']:,.0f} kg ({info['percentage']:.1f}%)" | |
| elements.append(Paragraph(summary_text, styles['Normal'])) | |
| elements.append(Spacer(1, 20)) | |
| elements.append(PageBreak()) | |
| elements.append(Paragraph(t['quality_control'], subtitle_style)) | |
| quality_data = [[t['material'], t['outliers'], t['normal_range_kg'], t['status']]] | |
| for material, info in outliers.items(): | |
| if info['count'] == 0: | |
| status = t['status_good'] | |
| elif info['count'] <= 3: | |
| status = t['status_monitor'] | |
| else: | |
| status = t['status_attention'] | |
| quality_data.append([ | |
| material.replace('_', ' ').title(), | |
| str(info['count']), | |
| info['range'], | |
| status | |
| ]) | |
| quality_table = Table(quality_data, colWidths=[2*inch, 1*inch, 2*inch, 1.5*inch]) | |
| quality_table.setStyle(TableStyle([ | |
| ('BACKGROUND', (0, 0), (-1, 0), colors.darkred), | |
| ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), | |
| ('ALIGN', (0, 0), (-1, -1), 'CENTER'), | |
| ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), | |
| ('GRID', (0, 0), (-1, -1), 1, colors.black), | |
| ('ROWBACKGROUNDS', (0, 1), (-1, -1), [colors.white, colors.lightgrey]) | |
| ])) | |
| elements.append(quality_table) | |
| if model: | |
| elements.append(PageBreak()) | |
| elements.append(Paragraph(t['intelligent_analysis'], subtitle_style)) | |
| try: | |
| analysis = generate_ai_summary(model, df, stats, outliers, lang) | |
| except: | |
| analysis = "Intelligent analysis temporarily unavailable." | |
| analysis_paragraphs = analysis.split('\n\n') | |
| for paragraph in analysis_paragraphs: | |
| if paragraph.strip(): | |
| formatted_text = paragraph.replace('**', '<b>', 1).replace('**', '</b>', 1) \ | |
| .replace('•', ' •') \ | |
| .replace('\n', '<br/>') | |
| elements.append(Paragraph(formatted_text, styles['Normal'])) | |
| elements.append(Spacer(1, 8)) | |
| elements.append(Spacer(1, 30)) | |
| footer_text = f""" | |
| <para alignment="center"> | |
| <i>{t['footer_generated']}<br/> | |
| {t['company']} - {t['division']}<br/> | |
| {t['footer_records'].format(records=f"{len(df):,}", days=stats['_total_']['work_days'])}</i> | |
| </para> | |
| """ | |
| elements.append(Paragraph(footer_text, styles['Normal'])) | |
| doc.build(elements) | |
| buffer.seek(0) | |
| return buffer | |
| def create_csv_export(df: pd.DataFrame, stats: Dict) -> pd.DataFrame: | |
| summary_df = pd.DataFrame([ | |
| { | |
| 'Material': material.replace('_', ' ').title(), | |
| 'Total_kg': info['total'], | |
| 'Percentage': info['percentage'], | |
| 'Daily_Average_kg': info['daily_avg'], | |
| 'Work_Days': info['work_days'], | |
| 'Records_Count': info['records'] | |
| } | |
| for material, info in stats.items() if material != '_total_' | |
| ]) | |
| return summary_df |