romfast/vending_data_intelligence_report
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: Data Intelligence Report Generator

Browse Source 
- Oracle ERP ROA integration with sales analytics and margin analysis
- Excel multi-sheet reports with conditional formatting
- PDF executive summaries with charts via ReportLab
- Optimized SQL queries (no cartesian products)
- Docker support for cross-platform deployment
- Configurable alert thresholds for business intelligence

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Marius Mutu
2025-12-02 15:41:56 +02:00
commit 0b732f7a7a
15 changed files with 5420 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
.env.example Normal file
View File

@@ -0,0 +1,15 @@
# Oracle connection settings
# ----------------------------
# For Windows native: use 127.0.0.1 or localhost
# For WSL: use Windows IP (run: cat /etc/resolv.conf | grep nameserver)
# For Docker: use host.docker.internal (automatic in docker-compose)
ORACLE_HOST=127.0.0.1
ORACLE_PORT=1521
ORACLE_SERVICE=XEPDB1
ORACLE_USER=FIRMA
ORACLE_PASSWORD=PAROLA
# Output settings
OUTPUT_DIR=./output
COMPANY_NAME=Data Intelligence Report

28
.gitignore vendored Normal file
View File

@@ -0,0 +1,28 @@
# Virtual environment
.venv/
venv/
ENV/
# Python cache
__pycache__/
*.py[cod]
*$py.class
*.so
# Environment variables (contains secrets)
.env
# Output files
output/
*.xlsx
*.pdf
# IDE
.vscode/
.idea/
*.swp
*.swo
# OS
.DS_Store
Thumbs.db

93
CLAUDE.md Normal file
View File

@@ -0,0 +1,93 @@
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## Project Overview
Data Intelligence Report Generator for ERP ROA (Oracle Database). Generates Excel and PDF business intelligence reports with sales analytics, margin analysis, stock tracking, and alerts.
## Commands
### Option 1: Virtual Environment (WSL or Windows)
```bash
# Create and activate virtual environment
python -m venv .venv
source .venv/bin/activate # Linux/WSL
# or: .venv\Scripts\activate # Windows
# Install dependencies
pip install -r requirements.txt
# Run report
python main.py
```
### Option 2: Docker (Windows Docker Desktop / Linux)
```bash
# Copy and configure environment
cp .env.example .env
# Edit .env with your Oracle credentials
# Run with docker-compose
docker-compose run --rm report-generator
# Or with custom months
docker-compose run --rm report-generator python main.py --months 6
```
### Common Options
```bash
# Run with custom period
python main.py --months 6
# Custom output directory
python main.py --output-dir /path/to/output
```
## Oracle Connection from Different Environments
| Environment | ORACLE_HOST value |
|-------------|-------------------|
| Windows native | `127.0.0.1` |
| WSL | Windows IP (run: `cat /etc/resolv.conf \| grep nameserver`) |
| Docker | `host.docker.internal` (automatic in docker-compose) |
## Architecture
**Entry point**: `main.py` - CLI interface, orchestrates query execution and report generation
**Data flow**:
1. `config.py` loads Oracle connection settings from `.env` file
2. `queries.py` contains all SQL queries in a `QUERIES` dictionary with metadata (title, description, params)
3. `main.py` executes queries via `OracleConnection` context manager, stores results in `results` dict
4. `report_generator.py` receives dataframes and generates:
- `ExcelReportGenerator`: Multi-sheet workbook with conditional formatting
- `PDFReportGenerator`: Executive summary with charts via ReportLab
**Key patterns**:
- Queries use parameterized `:months` for configurable analysis period
- Sheet order in `main.py:sheet_order` controls Excel tab sequence
- Charts are generated via matplotlib, converted to images for PDF
## Oracle Database Schema
Required views: `fact_vfacturi2`, `fact_vfacturi_detalii`, `vnom_articole`, `vnom_parteneri`, `vstoc`, `vrul`
Filter conventions:
- `sters = 0` excludes deleted records
- `tip NOT IN (7, 8, 9, 24)` excludes returns/credit notes
- Account codes: `341`, `345` = own production; `301` = raw materials
## Adding New Reports
1. Add SQL query constant in `queries.py`
2. Add entry to `QUERIES` dict with `sql`, `params`, `title`, `description`
3. Add query name to `sheet_order` list in `main.py` (line ~143)
4. For PDF inclusion, add rendering logic in `main.py:generate_reports()`
## Alert Thresholds (in config.py)
- Low margin: < 15%
- Price variation: > 20%
- Slow stock: > 90 days without movement
- Minimum sales for analysis: 1000 RON

17
Dockerfile Normal file
View File

@@ -0,0 +1,17 @@
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY *.py ./
COPY .env* ./
# Create output directory
RUN mkdir -p /app/output
# Default command
CMD ["python", "main.py"]

448
PLAN_FIXES_2025_11_28.md Normal file
View File

@@ -0,0 +1,448 @@
# Plan: Corectii Report Generator - 28.11.2025
## Probleme de Rezolvat
1. **Analiza Prajitorie** - intrarile si iesirile apar pe randuri diferite in loc de coloane
2. **Query-uri Financiare "No Data"** - DSO/DPO, Solduri clienti/furnizori, Aging, Pozitia Cash, Ciclu Conversie Cash nu afiseaza date (user confirma ca datele EXISTA)
3. **Recomandari in Sumar Executiv** - trebuie incluse sub KPIs in sheet-ul Sumar Executiv
4. **Reordonare Sheet-uri** - agregatele (indicatori_agregati, portofoliu_clienti, concentrare_risc) trebuie mutate imediat dupa Sumar Executiv
---
## ISSUE 1: Analiza Prajitorie - Restructurare din Randuri in Coloane
### Fisier: `queries.py` liniile 450-478
### Problema Curenta
Query-ul grupeaza dupa `tip_miscare` (Intrare/Iesire/Transformare), creand randuri separate:
```
luna | tip | tip_miscare | cantitate_intrata | cantitate_iesita
2024-01 | Materii prime | Intrare | 1000 | 0
2024-01 | Materii prime | Iesire | 0 | 800
```
### Output Cerut
Un rand per Luna + Tip cu coloane separate pentru Intrari si Iesiri:
```
luna | tip | cantitate_intrari | valoare_intrari | cantitate_iesiri | valoare_iesiri | sold_net
2024-01 | Materii prime | 1000 | 50000 | 800 | 40000 | 10000
```
### Solutia: Inlocuieste ANALIZA_PRAJITORIE (liniile 450-478)
```sql
ANALIZA_PRAJITORIE = """
SELECT
TO_CHAR(r.dataact, 'YYYY-MM') AS luna,
CASE
WHEN r.cont = '301' THEN 'Materii prime'
WHEN r.cont = '341' THEN 'Semifabricate'
WHEN r.cont = '345' THEN 'Produse finite'
ELSE 'Altele'
END AS tip,
-- Intrari: cantitate > 0 AND cante = 0
ROUND(SUM(CASE WHEN r.cant > 0 AND NVL(r.cante, 0) = 0 THEN r.cant ELSE 0 END), 2) AS cantitate_intrari,
ROUND(SUM(CASE WHEN r.cant > 0 AND NVL(r.cante, 0) = 0 THEN r.cant * NVL(r.pret, 0) ELSE 0 END), 2) AS valoare_intrari,
-- Iesiri: cant = 0 AND cante > 0
ROUND(SUM(CASE WHEN NVL(r.cant, 0) = 0 AND r.cante > 0 THEN r.cante ELSE 0 END), 2) AS cantitate_iesiri,
ROUND(SUM(CASE WHEN NVL(r.cant, 0) = 0 AND r.cante > 0 THEN r.cante * NVL(r.pret, 0) ELSE 0 END), 2) AS valoare_iesiri,
-- Transformari: cant > 0 AND cante > 0 (intrare si iesire simultan)
ROUND(SUM(CASE WHEN r.cant > 0 AND r.cante > 0 THEN r.cant ELSE 0 END), 2) AS cantitate_transformari_in,
ROUND(SUM(CASE WHEN r.cant > 0 AND r.cante > 0 THEN r.cante ELSE 0 END), 2) AS cantitate_transformari_out,
-- Sold net
ROUND(SUM(NVL(r.cant, 0) - NVL(r.cante, 0)), 2) AS sold_net_cantitate,
ROUND(SUM((NVL(r.cant, 0) - NVL(r.cante, 0)) * NVL(r.pret, 0)), 2) AS sold_net_valoare
FROM vrul r
WHERE r.cont IN ('301', '341', '345')
AND r.dataact >= ADD_MONTHS(TRUNC(SYSDATE), -:months)
GROUP BY TO_CHAR(r.dataact, 'YYYY-MM'),
CASE WHEN r.cont = '301' THEN 'Materii prime'
WHEN r.cont = '341' THEN 'Semifabricate'
WHEN r.cont = '345' THEN 'Produse finite'
ELSE 'Altele' END
ORDER BY luna, tip
"""
```
### Modificari Cheie
1. **Eliminat** `tip_miscare` din SELECT si GROUP BY
2. **Agregare conditionala** cu `CASE WHEN ... THEN ... ELSE 0 END` in SUM()
3. **Coloane separate** pentru fiecare tip de miscare
4. **Adaugat coloane valoare** pe langa cantitati
### Update Legends in main.py (in jurul liniei 224)
Adauga in dictionarul `legends`:
```python
'analiza_prajitorie': {
'CANTITATE_INTRARI': 'Cantitate intrata (cant > 0, cante = 0)',
'VALOARE_INTRARI': 'Valoare intrari = cantitate x pret',
'CANTITATE_IESIRI': 'Cantitate iesita (cant = 0, cante > 0)',
'VALOARE_IESIRI': 'Valoare iesiri = cantitate x pret',
'CANTITATE_TRANSFORMARI_IN': 'Cantitate intrata in transformari',
'CANTITATE_TRANSFORMARI_OUT': 'Cantitate iesita din transformari',
'SOLD_NET_CANTITATE': 'Sold net = Total intrari - Total iesiri',
'SOLD_NET_VALOARE': 'Valoare neta a soldului'
}
```
---
## ISSUE 2: Query-uri Financiare "No Data" - DIAGNOSTIC NECESAR
### Query-uri Afectate
| Query | View Folosit | Linie in queries.py | Filtru Curent |
|-------|--------------|---------------------|---------------|
| DSO_DPO | vbalanta_parteneri | 796-844 | `an = EXTRACT(YEAR FROM SYSDATE) AND luna = EXTRACT(MONTH FROM SYSDATE)` |
| SOLDURI_CLIENTI | vbalanta_parteneri | 636-654 | Acelasi + `cont LIKE '4111%'` |
| SOLDURI_FURNIZORI | vbalanta_parteneri | 659-677 | Acelasi + `cont LIKE '401%'` |
| AGING_CREANTE | vireg_parteneri | 682-714 | `cont LIKE '4111%' OR '461%'` |
| FACTURI_RESTANTE | vireg_parteneri | 719-734 | Acelasi + `datascad < SYSDATE` |
| POZITIA_CASH | vbal | 849-872 | `cont LIKE '512%' OR '531%'` |
| CICLU_CONVERSIE_CASH | Multiple | 877-940 | Combina toate de mai sus |
### User-ul confirma ca DATELE EXISTA - trebuie diagnosticat problema
### Cauze Posibile
1. Numele view-urilor difera in baza de date
2. Numele coloanelor difera (`an`, `luna`, `solddeb`, `soldcred`)
3. Prefixele codurilor de cont nu se potrivesc (4111%, 401%, 512%)
4. Pragurile HAVING sunt prea restrictive (`> 1`, `> 100`)
### FIX IMEDIAT: Relaxeaza Pragurile HAVING
**SOLDURI_CLIENTI** (linia 652):
```sql
-- DE LA:
HAVING ABS(SUM(b.solddeb - b.soldcred)) > 1
-- LA:
HAVING ABS(SUM(b.solddeb - b.soldcred)) > 0.01
```
**SOLDURI_FURNIZORI** (linia 675):
```sql
-- DE LA:
HAVING ABS(SUM(b.soldcred - b.solddeb)) > 1
-- LA:
HAVING ABS(SUM(b.soldcred - b.solddeb)) > 0.01
```
**AGING_CREANTE** (linia 712):
```sql
-- DE LA:
HAVING SUM(sold_ramas) > 100
-- LA:
HAVING SUM(sold_ramas) > 0.01
```
**AGING_DATORII** (linia 770):
```sql
-- DE LA:
HAVING SUM(sold_ramas) > 100
-- LA:
HAVING SUM(sold_ramas) > 0.01
```
**POZITIA_CASH** (linia 870):
```sql
-- DE LA:
HAVING ABS(SUM(b.solddeb - b.soldcred)) > 0.01
-- Deja OK, dar verifica daca vbal exista
```
### Daca Tot Nu Functioneaza - Verifica View-urile
Ruleaza in Oracle:
```sql
-- Verifica daca view-urile exista
SELECT view_name FROM user_views
WHERE view_name IN ('VBALANTA_PARTENERI', 'VIREG_PARTENERI', 'VBAL', 'VRUL');
-- Verifica daca exista date pentru luna curenta
SELECT an, luna, COUNT(*)
FROM vbalanta_parteneri
WHERE an = EXTRACT(YEAR FROM SYSDATE)
GROUP BY an, luna
ORDER BY luna DESC;
-- Verifica prefixele de cont existente
SELECT DISTINCT SUBSTR(cont, 1, 4) AS prefix_cont
FROM vbalanta_parteneri
WHERE an = EXTRACT(YEAR FROM SYSDATE);
```
---
## ISSUE 3: Recomandari in Sumar Executiv
### Stare Curenta
- Sheet `sumar_executiv` (linia 166) - contine doar KPIs
- Sheet `recomandari` (linia 168) - sheet separat cu toate recomandarile
### Solutia: Metoda noua in report_generator.py
### Adauga metoda noua in clasa `ExcelReportGenerator` (dupa linia 167 in report_generator.py)
```python
def add_sheet_with_recommendations(self, name: str, df: pd.DataFrame,
recommendations_df: pd.DataFrame,
title: str = None, description: str = None,
legend: dict = None, top_n_recommendations: int = 5):
"""Adauga sheet formatat cu KPIs si top recomandari dedesubt"""
sheet_name = name[:31]
ws = self.wb.create_sheet(title=sheet_name)
start_row = 1
# Adauga titlu
if title:
ws.cell(row=start_row, column=1, value=title)
ws.cell(row=start_row, column=1).font = Font(bold=True, size=14)
start_row += 1
# Adauga descriere
if description:
ws.cell(row=start_row, column=1, value=description)
ws.cell(row=start_row, column=1).font = Font(italic=True, size=10, color='666666')
start_row += 1
# Adauga timestamp
ws.cell(row=start_row, column=1, value=f"Generat: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
ws.cell(row=start_row, column=1).font = Font(size=9, color='999999')
start_row += 2
# === SECTIUNEA 1: KPIs ===
if df is not None and not df.empty:
# Header
for col_idx, col_name in enumerate(df.columns, 1):
cell = ws.cell(row=start_row, column=col_idx, value=col_name)
cell.font = self.header_font
cell.fill = self.header_fill
cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
cell.border = self.border
# Date
for row_idx, row in enumerate(df.itertuples(index=False), start_row + 1):
for col_idx, value in enumerate(row, 1):
cell = ws.cell(row=row_idx, column=col_idx, value=value)
cell.border = self.border
if isinstance(value, (int, float)):
cell.number_format = '#,##0.00' if isinstance(value, float) else '#,##0'
cell.alignment = Alignment(horizontal='right')
start_row = start_row + len(df) + 3
# === SECTIUNEA 2: TOP RECOMANDARI ===
if recommendations_df is not None and not recommendations_df.empty:
ws.cell(row=start_row, column=1, value="Top Recomandari Prioritare")
ws.cell(row=start_row, column=1).font = Font(bold=True, size=12, color='366092')
start_row += 1
# Sorteaza dupa prioritate (ALERTA primul, apoi ATENTIE, apoi OK)
df_sorted = recommendations_df.copy()
status_order = {'ALERTA': 0, 'ATENTIE': 1, 'OK': 2}
df_sorted['_order'] = df_sorted['STATUS'].map(status_order).fillna(3)
df_sorted = df_sorted.sort_values('_order').head(top_n_recommendations)
df_sorted = df_sorted.drop(columns=['_order'])
# Coloane de afisat
display_cols = ['STATUS', 'CATEGORIE', 'INDICATOR', 'VALOARE', 'RECOMANDARE']
display_cols = [c for c in display_cols if c in df_sorted.columns]
# Header cu background mov
for col_idx, col_name in enumerate(display_cols, 1):
cell = ws.cell(row=start_row, column=col_idx, value=col_name)
cell.font = self.header_font
cell.fill = PatternFill(start_color='8E44AD', end_color='8E44AD', fill_type='solid')
cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
cell.border = self.border
# Randuri cu colorare dupa status
for row_idx, (_, row) in enumerate(df_sorted.iterrows(), start_row + 1):
status = row.get('STATUS', 'OK')
for col_idx, col_name in enumerate(display_cols, 1):
value = row.get(col_name, '')
cell = ws.cell(row=row_idx, column=col_idx, value=value)
cell.border = self.border
cell.alignment = Alignment(wrap_text=True)
# Colorare condiționata
if status == 'ALERTA':
cell.fill = PatternFill(start_color='FADBD8', end_color='FADBD8', fill_type='solid')
elif status == 'ATENTIE':
cell.fill = PatternFill(start_color='FCF3CF', end_color='FCF3CF', fill_type='solid')
else:
cell.fill = PatternFill(start_color='D5F5E3', end_color='D5F5E3', fill_type='solid')
# Auto-adjust latime coloane
for col_idx in range(1, 8):
ws.column_dimensions[get_column_letter(col_idx)].width = 22
ws.freeze_panes = ws.cell(row=5, column=1)
```
### Modifica main.py - Loop-ul de Creare Sheet-uri (in jurul liniei 435)
```python
for query_name in sheet_order:
if query_name in results:
# Tratare speciala pentru 'sumar_executiv' - adauga recomandari sub KPIs
if query_name == 'sumar_executiv':
query_info = QUERIES.get(query_name, {})
excel_gen.add_sheet_with_recommendations(
name='Sumar Executiv',
df=results['sumar_executiv'],
recommendations_df=results.get('recomandari'),
title=query_info.get('title', 'Sumar Executiv'),
description=query_info.get('description', ''),
legend=legends.get('sumar_executiv'),
top_n_recommendations=5
)
# Pastreaza sheet-ul complet de recomandari
elif query_name == 'recomandari':
excel_gen.add_sheet(
name='RECOMANDARI',
df=results['recomandari'],
title='Recomandari Automate (Lista Completa)',
description='Toate insight-urile si actiunile sugerate bazate pe analiza datelor',
legend=legends.get('recomandari')
)
elif query_name in QUERIES:
# ... logica existenta neschimbata
```
---
## ISSUE 4: Reordonare Sheet-uri
### Fisier: `main.py` liniile 165-221
### Noul sheet_order (inlocuieste complet liniile 165-221)
```python
sheet_order = [
# SUMAR EXECUTIV
'sumar_executiv',
'sumar_executiv_yoy',
'recomandari',
# INDICATORI AGREGATI (MUTATI SUS - imagine de ansamblu)
'indicatori_agregati_venituri',
'indicatori_agregati_venituri_yoy',
'portofoliu_clienti',
'concentrare_risc',
'concentrare_risc_yoy',
'sezonalitate_lunara',
# INDICATORI GENERALI & LICHIDITATE
'indicatori_generali',
'indicatori_lichiditate',
'clasificare_datorii',
'grad_acoperire_datorii',
'proiectie_lichiditate',
# ALERTE
'vanzari_sub_cost',
'clienti_marja_mica',
# CICLU CASH
'ciclu_conversie_cash',
# ANALIZA CLIENTI
'marja_per_client',
'clienti_ranking_profit',
'frecventa_clienti',
'concentrare_clienti',
'trending_clienti',
'marja_client_categorie',
# PRODUSE
'top_produse',
'marja_per_categorie',
'marja_per_gestiune',
'articole_negestionabile',
'productie_vs_revanzare',
# PRETURI
'dispersie_preturi',
'clienti_sub_medie',
'evolutie_discount',
# FINANCIAR
'dso_dpo',
'dso_dpo_yoy',
'solduri_clienti',
'aging_creante',
'facturi_restante',
'solduri_furnizori',
'aging_datorii',
'facturi_restante_furnizori',
'pozitia_cash',
# ISTORIC
'vanzari_lunare',
# STOC
'stoc_curent',
'stoc_lent',
'rotatie_stocuri',
# PRODUCTIE
'analiza_prajitorie',
]
```
---
## Ordinea de Implementare
### Pasul 1: queries.py
1. Inlocuieste ANALIZA_PRAJITORIE (liniile 450-478) cu versiunea cu agregare conditionala
2. Relaxeaza pragurile HAVING in:
- SOLDURI_CLIENTI (linia 652): `> 1` -> `> 0.01`
- SOLDURI_FURNIZORI (linia 675): `> 1` -> `> 0.01`
- AGING_CREANTE (linia 712): `> 100` -> `> 0.01`
- AGING_DATORII (linia 770): `> 100` -> `> 0.01`
### Pasul 2: report_generator.py
1. Adauga metoda `add_sheet_with_recommendations()` dupa linia 167
2. Asigura-te ca importurile includ `PatternFill`, `get_column_letter` din openpyxl
### Pasul 3: main.py
1. Inlocuieste array-ul `sheet_order` (liniile 165-221)
2. Modifica loop-ul de creare sheet-uri pentru `sumar_executiv` (in jurul liniei 435)
3. Adauga legend pentru `analiza_prajitorie` in dictionarul `legends`
### Pasul 4: Testare
1. Ruleaza cu `python main.py --months 1` pentru test rapid
2. Verifica sheet-ul `analiza_prajitorie` - format columnar
3. Verifica query-urile financiare - trebuie sa returneze date
4. Verifica `Sumar Executiv` - sectiune recomandari dedesubt
5. Verifica ordinea sheet-urilor - agregatele dupa sumar
---
## Fisiere Critice
| Fisier | Ce se modifica | Linii |
|--------|----------------|-------|
| `queries.py` | ANALIZA_PRAJITORIE SQL | 450-478 |
| `queries.py` | HAVING thresholds | 652, 675, 712, 770 |
| `report_generator.py` | Metoda noua | dupa 167 |
| `main.py` | sheet_order array | 165-221 |
| `main.py` | Loop creare sheet-uri | ~435 |
| `main.py` | legends dict | ~224 |
---
## Note pentru Sesiunea Viitoare
1. **Prioritate ALERTA**: Query-urile financiare "no data" - user-ul a confirmat ca datele EXISTA. Daca relaxarea HAVING nu rezolva, trebuie verificate numele view-urilor si coloanelor in Oracle.
2. **Import necesar** in report_generator.py:
```python
from openpyxl.utils import get_column_letter
from openpyxl.styles import PatternFill
```
3. **Testare**: Dupa implementare, ruleaza raportul si verifica fiecare din cele 4 fix-uri.

176
README.md Normal file
View File

@@ -0,0 +1,176 @@
# Data Intelligence Report Generator
Generator automat de rapoarte de Business Intelligence pentru ERP ROA (Oracle Database).
## 📋 Funcționalități
Scriptul generează automat:
### Excel (multiple sheet-uri):
- **Sumar Executiv** - KPIs principale
- **🚨 Vânzări sub Cost** - Alertă critică pentru pierderi
- **⚠️ Clienți Marjă Mică** - Clienți care necesită renegociere
- **Marja per Client** - Profitabilitate per client
- **Concentrare Clienți** - Analiza dependenței de clienți mari
- **Trending Clienți** - Evoluție YoY (creștere/scădere)
- **Top Produse** - Cele mai vândute produse
- **Marja per Categorie** - Profitabilitate per subgrupă
- **Marja per Gestiune** - Profitabilitate per depozit
- **Producție vs Revânzare** - Comparație marje
- **Dispersie Prețuri** - Produse cu variație mare de preț
- **Clienți sub Medie** - Cine cumpără prea ieftin
- **Vânzări Lunare** - Sezonalitate (24 luni)
- **Stoc Curent** - Valoare stoc per gestiune
- **Stoc Lent** - Produse fără mișcare >90 zile
- **Rotație Stocuri** - Viteza de rotație
- **Analiză Prăjitorie** - Fluxul producției
### PDF (Sumar Executiv):
- Grafice interactive
- Tabele rezumat
- Alerte evidențiate vizual
- Perfect pentru prezentări
## 🚀 Instalare
### Opțiunea 1: Virtual Environment (recomandat)
```bash
# 1. Clonează/copiază directorul
cd data_intelligence_report
# 2. Creează mediu virtual
python -m venv .venv
# 3. Activează mediul virtual
source .venv/bin/activate # Linux/WSL/Mac
# sau
.venv\Scripts\activate # Windows CMD
# sau
.venv\Scripts\Activate.ps1 # Windows PowerShell
# 4. Instalează dependențele
pip install -r requirements.txt
# 5. Configurează conexiunea
cp .env.example .env
nano .env # sau editează cu alt editor
```
### Opțiunea 2: Docker
```bash
# 1. Configurează conexiunea
cp .env.example .env
nano .env
# 2. Rulează cu docker-compose
docker-compose run --rm report-generator
```
## ⚙️ Configurare
Copiază `.env.example` în `.env` și editează:
```env
ORACLE_HOST=127.0.0.1
ORACLE_PORT=1521
ORACLE_SERVICE=XEPDB1
ORACLE_USER=FIRMA
ORACLE_PASSWORD=PAROLA
OUTPUT_DIR=./output
COMPANY_NAME=Numele Companiei
```
### Conexiune Oracle din diferite medii
| Mediu | ORACLE_HOST |
|-------|-------------|
| Windows nativ | `127.0.0.1` sau `localhost` |
| WSL | IP Windows (vezi mai jos) |
| Docker | `host.docker.internal` (automat în docker-compose) |
**Pentru WSL**, obține IP-ul Windows:
```bash
cat /etc/resolv.conf | grep nameserver | awk '{print $2}'
```
## 📊 Utilizare
### Cu Virtual Environment
```bash
# Activează mediul (dacă nu e activ)
source .venv/bin/activate # Linux/WSL
# sau: .venv\Scripts\activate # Windows
# Raport standard (ultimele 12 luni)
python main.py
# Raport pentru ultimele 6 luni
python main.py --months 6
# Salvare în alt director
python main.py --output-dir /path/to/output
```
### Cu Docker
```bash
# Raport standard
docker-compose run --rm report-generator
# Cu parametri custom
docker-compose run --rm report-generator python main.py --months 6
```
## 📁 Structura Output
```
output/
├── data_intelligence_report_20251126_143022.xlsx
└── data_intelligence_report_20251126_143022.pdf
```
## 🔧 Cerințe Sistem
- Python 3.8+
- Oracle Client (sau oracledb în thin mode)
- Acces la schema Oracle cu view-urile:
- `fact_vfacturi2`
- `fact_vfacturi_detalii`
- `vnom_articole`
- `vnom_parteneri`
- `vstoc`
- `vrul`
## 📈 Query-uri Incluse
Toate query-urile sunt în `queries.py` și pot fi personalizate:
1. **Praguri de alertă**:
- Marjă mică: < 15%
- Variație preț: > 20%
- Stoc lent: > 90 zile
2. **Parametri configurabili**:
- Perioada de analiză (luni)
- Vânzări minime pentru analiză
- Număr maxim de rezultate
## 🤝 Personalizare
Pentru a adăuga query-uri noi:
1. Adaugă SQL-ul în `queries.py`
2. Adaugă entry în dicționarul `QUERIES`
3. (Opțional) Adaugă în `sheet_order` din `main.py`
## 📝 Note
- Raportul folosește view-urile existente din ERP
- Datele șterse (sters=0) sunt excluse automat
- Retururile și credit note-urile sunt excluse din analiză
- Producția proprie e identificată după cont (341, 345)
---
Generator creat pentru ROMFAST SRL - ERP ROA

85
config.py Normal file
View File

@@ -0,0 +1,85 @@
"""
Configuration module - reads settings from .env file
"""
import os
from pathlib import Path
from dotenv import load_dotenv
# Load .env file from the same directory as this script
env_path = Path(__file__).parent / '.env'
load_dotenv(env_path)
# Oracle connection settings
ORACLE_CONFIG = {
'host': os.getenv('ORACLE_HOST', '127.0.0.1'),
'port': os.getenv('ORACLE_PORT', '1521'),
'service': os.getenv('ORACLE_SERVICE', 'XEPDB1'),
'user': os.getenv('ORACLE_USER', 'FIRMA'),
'password': os.getenv('ORACLE_PASSWORD', 'PAROLA'),
}
# Build DSN string
def get_dsn():
return f"{ORACLE_CONFIG['host']}:{ORACLE_CONFIG['port']}/{ORACLE_CONFIG['service']}"
# Output settings
OUTPUT_DIR = Path(os.getenv('OUTPUT_DIR', './output'))
COMPANY_NAME = os.getenv('COMPANY_NAME', 'Data Intelligence Report')
# Ensure output directory exists
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# Analysis parameters
ANALYSIS_MONTHS = 12 # Default: last 12 months
MIN_SALES_FOR_ANALYSIS = 1000 # Minimum RON to include client in analysis
LOW_MARGIN_THRESHOLD = 15 # Alert if margin below 15%
PRICE_VARIATION_THRESHOLD = 20 # Alert if price varies > 20% for same product
# Recommendation thresholds for business intelligence
RECOMMENDATION_THRESHOLDS = {
# Margin thresholds
'marja_minima': 15, # % - below this = problem
'marja_tinta': 20, # % - target margin
# Client concentration risk
'concentrare_top5_max': 60, # % - risk if top 5 clients > 60%
'concentrare_top1_max': 25, # % - risk if single client > 25%
# Financial indicators
'dso_target': 45, # days - DSO target
'dso_alert': 60, # days - DSO alert threshold
'dpo_target': 30, # days - DPO target
# Stock indicators
'stoc_zile_max': 90, # days - slow stock threshold
'rotatie_minima': 4, # minimum annual rotation
# Receivables aging
'restante_90_procent': 10, # % - max receivables > 90 days
# ==========================================================================
# NEW: Indicatori generali de business (PLAN_INDICATORI_LICHIDITATE_YOY)
# ==========================================================================
'grad_indatorare_max': 1.0, # Datorii/Capital propriu - OK sub 1
'grad_indatorare_alert': 2.0, # Alertă peste 2.0
'autonomie_financiara_min': 0.5, # Capital propriu/Activ total - OK peste 0.5
'autonomie_financiara_alert': 0.3, # Alertă sub 0.3
'rata_datoriilor_max': 0.5, # Datorii/Activ total - OK sub 0.5
'rata_datoriilor_alert': 0.7, # Alertă peste 0.7
'marja_neta_min': 5, # % - Profit/Vânzări - OK peste 5%
'marja_neta_alert': 3, # Alertă sub 3%
'roa_min': 5, # % - Profit/Activ total - OK peste 5%
'roa_alert': 2, # Alertă sub 2%
'rotatia_activelor_min': 1.0, # Vânzări/Activ - OK peste 1.0
# ==========================================================================
# NEW: Indicatori de lichiditate
# ==========================================================================
'lichiditate_curenta_min': 1.0, # (Cash+Creanțe+Stoc)/Datorii curente - minim
'lichiditate_curenta_target': 1.5, # Ținta ideală
'lichiditate_rapida_min': 0.7, # (Cash+Creanțe)/Datorii curente - minim
'lichiditate_rapida_target': 1.0, # Ținta ideală
'cash_ratio_min': 0.1, # Cash/Datorii curente - minim
'cash_ratio_target': 0.2, # Ținta ideală
'grad_acoperire_min': 1.0, # (Cash+Încasări 30z)/Datorii 30z - minim
}

19
docker-compose.yml Normal file
View File

@@ -0,0 +1,19 @@
services:
report-generator:
build: .
volumes:
# Mount output directory for generated reports
- ./output:/app/output
# Mount .env file (create from .env.example)
- ./.env:/app/.env:ro
environment:
# For Docker Desktop on Windows: host.docker.internal resolves to Windows host
# Override ORACLE_HOST if needed
- ORACLE_HOST=${ORACLE_HOST:-host.docker.internal}
- ORACLE_PORT=${ORACLE_PORT:-1521}
- ORACLE_SERVICE=${ORACLE_SERVICE:-XEPDB1}
- ORACLE_USER=${ORACLE_USER:-FIRMA}
- ORACLE_PASSWORD=${ORACLE_PASSWORD:-PAROLA}
# Add host.docker.internal for Linux Docker (not needed on Windows/Mac)
extra_hosts:
- "host.docker.internal:host-gateway"

670
main.py Normal file
View File

@@ -0,0 +1,670 @@
#!/usr/bin/env python3
"""
Data Intelligence Report Generator
===================================
Generates comprehensive Excel and PDF reports with financial analytics
for ERP data stored in Oracle Database.
Usage:
python main.py [--months 12] [--output-dir ./output]
Author: Claude AI for ROMFAST SRL
"""
import sys
import argparse
from datetime import datetime
from pathlib import Path
import warnings
warnings.filterwarnings('ignore')
# Check dependencies
def check_dependencies():
"""Check if all required packages are installed"""
missing = []
packages = {
'oracledb': 'oracledb',
'pandas': 'pandas',
'openpyxl': 'openpyxl',
'matplotlib': 'matplotlib',
'reportlab': 'reportlab',
'dotenv': 'python-dotenv'
}
for import_name, pip_name in packages.items():
try:
__import__(import_name)
except ImportError:
missing.append(pip_name)
if missing:
print("❌ Pachete lipsă. Rulează:")
print(f" pip install {' '.join(missing)} --break-system-packages")
sys.exit(1)
check_dependencies()
import oracledb
import pandas as pd
from dateutil.relativedelta import relativedelta
from config import (
ORACLE_CONFIG, get_dsn, OUTPUT_DIR, COMPANY_NAME,
ANALYSIS_MONTHS, MIN_SALES_FOR_ANALYSIS, LOW_MARGIN_THRESHOLD,
RECOMMENDATION_THRESHOLDS
)
from queries import QUERIES
from report_generator import (
ExcelReportGenerator, PDFReportGenerator,
create_monthly_chart, create_client_concentration_chart, create_production_chart,
create_cash_cycle_chart
)
from recommendations import RecommendationsEngine
class OracleConnection:
"""Context manager for Oracle database connection"""
def __init__(self):
self.connection = None
def __enter__(self):
try:
print(f"🔌 Conectare la Oracle: {get_dsn()}...")
self.connection = oracledb.connect(
user=ORACLE_CONFIG['user'],
password=ORACLE_CONFIG['password'],
dsn=get_dsn()
)
print("✓ Conectat cu succes!")
return self.connection
except oracledb.Error as e:
print(f"❌ Eroare conexiune Oracle: {e}")
raise
def __exit__(self, exc_type, exc_val, exc_tb):
if self.connection:
self.connection.close()
print("✓ Conexiune închisă.")
def execute_query(connection, query_name: str, query_info: dict) -> pd.DataFrame:
"""Execute a query and return results as DataFrame"""
try:
sql = query_info['sql']
params = query_info.get('params', {})
print(f" 📊 Executare: {query_name}...", end=" ")
with connection.cursor() as cursor:
cursor.execute(sql, params)
columns = [col[0] for col in cursor.description]
rows = cursor.fetchall()
df = pd.DataFrame(rows, columns=columns)
print(f"✓ ({len(df)} rânduri)")
return df
except oracledb.Error as e:
print(f"❌ Eroare: {e}")
return pd.DataFrame()
def generate_reports(args):
"""Main function to generate all reports"""
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
excel_path = OUTPUT_DIR / f"data_intelligence_report_{timestamp}.xlsx"
pdf_path = OUTPUT_DIR / f"data_intelligence_report_{timestamp}.pdf"
print("\n" + "="*60)
print(" DATA INTELLIGENCE REPORT GENERATOR")
print("="*60)
print(f" Perioada: Ultimele {args.months} luni")
print(f" Output: {OUTPUT_DIR}")
print("="*60 + "\n")
# Update parameters with command line arguments
for query_info in QUERIES.values():
if 'months' in query_info.get('params', {}):
query_info['params']['months'] = args.months
# Calculate reporting period string
end_date = datetime.now()
start_date = end_date - relativedelta(months=args.months)
period_str = f"Perioada: {start_date.strftime('%d.%m.%Y')} - {end_date.strftime('%d.%m.%Y')}"
# Add period to descriptions for queries with months parameter
for query_name, query_info in QUERIES.items():
if 'months' in query_info.get('params', {}):
original_desc = query_info.get('description', '')
query_info['description'] = f"{original_desc}\n{period_str}"
# Connect and execute queries
results = {}
with OracleConnection() as conn:
print("\n📥 Extragere date din Oracle:\n")
for query_name, query_info in QUERIES.items():
df = execute_query(conn, query_name, query_info)
results[query_name] = df
# Generate Excel Report
print("\n📝 Generare raport Excel...")
excel_gen = ExcelReportGenerator(excel_path)
# Generate recommendations based on all data
print("\n🔍 Generare recomandări automate...")
recommendations_engine = RecommendationsEngine(RECOMMENDATION_THRESHOLDS)
recommendations_df = recommendations_engine.analyze_all(results)
results['recomandari'] = recommendations_df
print(f"✓ {len(recommendations_df)} recomandări generate")
# Add sheets in logical order (updated per PLAN_INDICATORI_LICHIDITATE_YOY.md)
sheet_order = [
# SUMAR EXECUTIV
'sumar_executiv',
'sumar_executiv_yoy',
'recomandari',
# INDICATORI AGREGATI (MUTATI SUS - imagine de ansamblu)
'indicatori_agregati_venituri',
'indicatori_agregati_venituri_yoy',
'portofoliu_clienti',
'concentrare_risc',
'concentrare_risc_yoy',
'sezonalitate_lunara',
# INDICATORI GENERALI & LICHIDITATE
'indicatori_generali',
'indicatori_lichiditate',
'clasificare_datorii',
'grad_acoperire_datorii',
'proiectie_lichiditate',
# ALERTE
'vanzari_sub_cost',
'clienti_marja_mica',
# CICLU CASH
'ciclu_conversie_cash',
# ANALIZA CLIENTI
'marja_per_client',
'clienti_ranking_profit',
'frecventa_clienti',
'concentrare_clienti',
'trending_clienti',
'marja_client_categorie',
# PRODUSE
'top_produse',
'marja_per_categorie',
'marja_per_gestiune',
'articole_negestionabile',
'productie_vs_revanzare',
# PRETURI
'dispersie_preturi',
'clienti_sub_medie',
'evolutie_discount',
# FINANCIAR
'dso_dpo',
'dso_dpo_yoy',
'solduri_clienti',
'aging_creante',
'facturi_restante',
'solduri_furnizori',
'aging_datorii',
'facturi_restante_furnizori',
'pozitia_cash',
# ISTORIC
'vanzari_lunare',
# STOC
'stoc_curent',
'stoc_lent',
'rotatie_stocuri',
# PRODUCTIE
'analiza_prajitorie',
]
# Legends for each sheet explaining column calculations
legends = {
'recomandari': {
'CATEGORIE': 'Domeniu: Marja, Clienti, Stoc, Financiar',
'STATUS': 'OK = bine, ATENTIE = necesita atentie, ALERTA = actiune urgenta',
'VEZI_DETALII': 'Sheet-ul cu date detaliate pentru acest indicator'
},
'marja_per_client': {
'VANZARI_FARA_TVA': 'SUM(cantitate × preț) din fact_vfacturi_detalii',
'COST_TOTAL': 'SUM(cantitate × pret_achizitie)',
'MARJA_BRUTA': 'Vânzări - Cost = SUM(cantitate × (preț - pret_achizitie))',
'PROCENT_MARJA': 'Marja Brută / Vânzări × 100'
},
'clienti_marja_mica': {
'VANZARI_FARA_TVA': 'SUM(cantitate × preț) pentru client',
'MARJA_BRUTA': 'Vânzări - Cost',
'PROCENT_MARJA': 'Marja / Vânzări × 100 (sub 15% = alertă)'
},
'vanzari_sub_cost': {
'PRET_VANZARE': 'Preț unitar din factură (fact_vfacturi_detalii.pret)',
'COST': 'Preț achiziție (pret_achizitie)',
'PIERDERE': '(Preț vânzare - Cost) × Cantitate (negativ = pierdere)'
},
'stoc_curent': {
'TIP_GESTIUNE': 'Preț vânzare (nr_pag=7) sau Preț achiziție',
'VALOARE_STOC_ACHIZITIE': '(cants+cant-cante) × pret din vstoc',
'VALOARE_STOC_VANZARE': 'Doar pentru gestiuni preț vânzare, altfel gol'
},
'stoc_lent': {
'CANTITATE': 'Stoc final = cants + cant - cante',
'VALOARE': 'Cantitate × preț achiziție',
'ZILE_FARA_MISCARE': 'Zile de la ultima ieșire (dataout) sau intrare'
},
'rotatie_stocuri': {
'VALOARE_STOC': 'Stoc curent (cants+cant-cante) × preț achiziție',
'VANZARI_12_LUNI': 'Doar vânzări (nu transferuri/consumuri) din ultimele 12 luni',
'ROTATIE': 'Vânzări / Stoc (de câte ori s-a rotit stocul)',
'ZILE_STOC': 'La ritmul actual, în câte zile se epuizează'
},
'dispersie_preturi': {
'NR_TRANZACTII': 'Număr total linii factură pentru acest produs',
'VARIATIE_PROCENT': '(Preț Max - Preț Min) / Preț Mediu × 100',
'NR_LA_PRET_MIN': 'Câte tranzacții au fost la prețul minim',
'CLIENT_PRET_MIN': 'Primul client care a cumpărat la preț minim'
},
'top_produse': {
'VALOARE_VANZARI': 'SUM(cantitate × preț)',
'MARJA_BRUTA': 'SUM(cantitate × (preț - pret_achizitie))',
'PROCENT_MARJA': 'Marja / Vânzări × 100'
},
'marja_per_categorie': {
'VANZARI_FARA_TVA': 'Total vânzări pe subgrupă',
'COST_TOTAL': 'Total cost achiziție pe subgrupă',
'PROCENT_MARJA': 'Marja / Vânzări × 100'
},
'marja_per_gestiune': {
'VANZARI_FARA_TVA': 'Total vânzări pe gestiune (doar articole gestionabile)',
'MARJA_BRUTA': 'Total marjă pe gestiune',
'PROCENT_MARJA': 'Marja / Vânzări × 100'
},
'articole_negestionabile': {
'DENUMIRE': 'Nume articol negestionabil (in_stoc=0)',
'VANZARI_FARA_TVA': 'Total vânzări pentru articole care nu se țin pe stoc',
'MARJA_BRUTA': 'Vânzări - Cost',
'PROCENT_MARJA': 'Marja / Vânzări × 100'
},
'vanzari_lunare': {
'VANZARI_FARA_TVA': 'Total vânzări în lună',
'MARJA_BRUTA': 'Total marjă în lună',
'NR_FACTURI': 'Număr facturi emise',
'NR_CLIENTI': 'Clienți unici activi'
},
# NEW legends for financial and aggregated sheets
'indicatori_agregati_venituri': {
'LINIE_BUSINESS': 'Producție proprie / Materii prime / Marfă revândută',
'PROCENT_VENITURI': 'Contribuția la totalul vânzărilor',
'CONTRIBUTIE_PROFIT': 'Contribuția la profitul total (%)'
},
'sezonalitate_lunara': {
'MEDIE_VANZARI': 'Media vânzărilor pe 24 luni pentru această lună',
'DEVIERE_PROCENT': 'Cât de mult deviază de la media globală',
'CLASIFICARE': 'LUNĂ PUTERNICĂ / LUNĂ SLABĂ / NORMAL'
},
'portofoliu_clienti': {
'VALOARE': 'Numărul de clienți în fiecare categorie',
'EXPLICATIE': 'Definiția categoriei de clienți'
},
'concentrare_risc': {
'PROCENT': 'Procentul din vânzări pentru Top N clienți',
'STATUS': 'OK / ATENTIE / RISC MARE'
},
'ciclu_conversie_cash': {
'INDICATOR': 'DIO (zile stoc) + DSO (zile încasare) - DPO (zile plată)',
'ZILE': 'Numărul de zile pentru fiecare component',
'EXPLICATIE': 'Ce reprezintă fiecare indicator'
},
'clienti_ranking_profit': {
'RANG_PROFIT': 'Poziția clientului după profit (nu vânzări)',
'RANG_VANZARI': 'Poziția clientului după vânzări',
'PROFIT_BRUT': 'Vânzări - Cost = profitul efectiv adus'
},
'frecventa_clienti': {
'COMENZI_PE_LUNA': 'Media comenzilor pe lună',
'VALOARE_MEDIE_COMANDA': 'Valoarea medie per comandă',
'EVOLUTIE_FRECVENTA_YOY': 'Schimbarea frecvenței față de anul trecut'
},
'marja_client_categorie': {
'STATUS_MARJA': 'OK / MARJĂ MICĂ (<15%) / PIERDERE (negativă)',
'CATEGORIA': 'Grupa de produse',
'PROCENT_MARJA': 'Marja pentru acest client la această categorie'
},
'evolutie_discount': {
'PRET_INITIAL': 'Prețul mediu în primele 6 luni',
'PRET_ACTUAL': 'Prețul mediu în ultimele 6 luni',
'VARIATIE_PRET_PROCENT': 'Scăderea/creșterea prețului (negativ = discount)'
},
'dso_dpo': {
'DSO': 'Days Sales Outstanding - zile medii încasare clienți',
'DPO': 'Days Payables Outstanding - zile medii plată furnizori',
'STATUS': 'OK / ATENTIE / ALERTA'
},
'solduri_clienti': {
'SOLD_CURENT': 'Suma de încasat de la client (din cont 4111)',
'TIP_SOLD': 'Creanță (ne datorează) sau Avans client (am încasat în avans)'
},
'aging_creante': {
'NEAJUNS_SCADENTA': 'Facturi nescadente încă',
'ZILE_1_30': 'Restanțe 1-30 zile',
'PESTE_90_ZILE': 'Restanțe critice >90 zile - risc de neîncasare'
},
'facturi_restante': {
'ZILE_INTARZIERE': 'Zile de la scadență',
'SUMA_RESTANTA': 'Valoarea rămasă de încasat'
},
'aging_datorii': {
'NEAJUNS_SCADENTA': 'Datorii neajunse la scadență',
'ZILE_1_30': 'Restanțe 1-30 zile',
'ZILE_31_60': 'Restanțe 31-60 zile',
'ZILE_61_90': 'Restanțe 61-90 zile',
'PESTE_90_ZILE': 'Restanțe critice >90 zile',
'TOTAL_SOLD': 'Total datorii către furnizor'
},
'facturi_restante_furnizori': {
'ZILE_INTARZIERE': 'Zile de la scadență',
'SUMA_RESTANTA': 'Valoarea rămasă de plătit'
},
'solduri_furnizori': {
'SOLD_CURENT': 'Suma de plătit furnizorului (din cont 401)',
'TIP_SOLD': 'Datorie (trebuie să plătim) sau Avans (am plătit în avans)'
},
'pozitia_cash': {
'SOLD_CURENT': 'Disponibilul curent în cont/casă',
'DESCRIERE': 'Tipul contului (bancă/casă, lei/valută)'
},
# =====================================================================
# NEW: Legends for Indicatori Generali, Lichiditate, YoY sheets
# =====================================================================
'indicatori_generali': {
'INDICATOR': 'Grad îndatorare, autonomie financiară, ROA, marjă netă',
'VALOARE': 'Valoarea calculată a indicatorului',
'STATUS': 'OK / ATENȚIE / ALERTĂ bazat pe praguri standard',
'RECOMANDARE': 'Acțiune sugerată pentru îmbunătățire'
},
'indicatori_lichiditate': {
'INDICATOR': 'Lichiditate curentă, rapidă, cash ratio, fond de rulment',
'VALOARE': 'Valoarea calculată (rată sau sumă RON)',
'STATUS': 'OK / ATENȚIE / ALERTĂ',
'INTERPRETARE': 'Ce înseamnă valoarea pentru business'
},
'clasificare_datorii': {
'CATEGORIE': 'Termen scurt (<30z) / mediu (31-90z) / lung (>90z)',
'VALOARE': 'Suma datoriilor în categoria respectivă',
'NR_FACTURI': 'Numărul facturilor în acea categorie'
},
'grad_acoperire_datorii': {
'VALOARE': 'Cash disponibil + încasări așteptate vs plăți scadente',
'ACOPERIRE': 'OK / ATENȚIE / DEFICIT - dacă puteți plăti datoriile',
'EXPLICATIE': 'Ce înseamnă pentru fluxul de numerar'
},
'proiectie_lichiditate': {
'PERIOADA': 'Azi / 30 zile / 60 zile / 90 zile',
'SOLD_PROIECTAT': 'Cash estimat la sfârșitul perioadei',
'FLUX_NET': 'Încasări - Plăți pentru perioada respectivă',
'STATUS': 'OK dacă sold pozitiv, ALERTĂ dacă negativ'
},
'sumar_executiv_yoy': {
'VALOARE_CURENTA': 'Valoarea din ultimele 12 luni',
'VALOARE_ANTERIOARA': 'Valoarea din anul anterior (12-24 luni)',
'VARIATIE_PROCENT': 'Creștere/scădere procentuală',
'TREND': 'CREȘTERE / SCĂDERE / STABIL'
},
'dso_dpo_yoy': {
'VALOARE_CURENTA': 'Zile încasare/plată actuale',
'VALOARE_ANTERIOARA': 'Zile în perioada anterioară',
'VARIATIE_ZILE': 'Diferența în zile (+ = mai rău pentru DSO, mai bine pentru DPO)',
'TREND': 'ÎMBUNĂTĂȚIRE / DETERIORARE / STABIL'
},
'concentrare_risc_yoy': {
'PROCENT_CURENT': '% vânzări la Top N clienți - an curent',
'PROCENT_ANTERIOR': '% vânzări la Top N clienți - an trecut',
'VARIATIE': 'Schimbarea în puncte procentuale',
'TREND': 'DIVERSIFICARE (bine) / CONCENTRARE (risc) / STABIL'
},
'indicatori_agregati_venituri_yoy': {
'LINIE_BUSINESS': 'Producție proprie / Materii prime / Marfă',
'VANZARI_CURENTE': 'Vânzări în ultimele 12 luni',
'VANZARI_ANTERIOARE': 'Vânzări în perioada anterioară',
'VARIATIE_PROCENT': 'Creștere/scădere procentuală',
'TREND': 'CREȘTERE / SCĂDERE / STABIL'
},
'analiza_prajitorie': {
'CANTITATE_INTRARI': 'Cantitate intrata (cant > 0, cante = 0)',
'VALOARE_INTRARI': 'Valoare intrari = cantitate x pret',
'CANTITATE_IESIRI': 'Cantitate iesita (cant = 0, cante > 0)',
'VALOARE_IESIRI': 'Valoare iesiri = cantitate x pret',
'CANTITATE_TRANSFORMARI_IN': 'Cantitate intrata in transformari',
'CANTITATE_TRANSFORMARI_OUT': 'Cantitate iesita din transformari',
'SOLD_NET_CANTITATE': 'Sold net = Total intrari - Total iesiri',
'SOLD_NET_VALOARE': 'Valoare neta a soldului'
}
}
for query_name in sheet_order:
if query_name in results:
# Tratare speciala pentru 'sumar_executiv' - adauga recomandari sub KPIs
if query_name == 'sumar_executiv':
query_info = QUERIES.get(query_name, {})
excel_gen.add_sheet_with_recommendations(
name='Sumar Executiv',
df=results['sumar_executiv'],
recommendations_df=results.get('recomandari'),
title=query_info.get('title', 'Sumar Executiv'),
description=query_info.get('description', ''),
legend=legends.get('sumar_executiv'),
top_n_recommendations=5
)
# Pastreaza sheet-ul complet de recomandari
elif query_name == 'recomandari':
excel_gen.add_sheet(
name='RECOMANDARI',
df=results['recomandari'],
title='Recomandari Automate (Lista Completa)',
description='Toate insight-urile si actiunile sugerate bazate pe analiza datelor',
legend=legends.get('recomandari')
)
elif query_name in QUERIES:
query_info = QUERIES[query_name]
# Create short sheet name from query name
sheet_name = query_name.replace('_', ' ').title()[:31]
excel_gen.add_sheet(
name=sheet_name,
df=results[query_name],
title=query_info.get('title', query_name),
description=query_info.get('description', ''),
legend=legends.get(query_name)
)
excel_gen.save()
# Generate PDF Report
print("\n📄 Generare raport PDF...")
pdf_gen = PDFReportGenerator(pdf_path, company_name=COMPANY_NAME)
# Title page
pdf_gen.add_title_page()
# KPIs
pdf_gen.add_kpi_section(results.get('sumar_executiv'))
# NEW: Indicatori Generali section
if 'indicatori_generali' in results and not results['indicatori_generali'].empty:
pdf_gen.add_table_section(
"Indicatori Generali de Business",
results.get('indicatori_generali'),
columns=['INDICATOR', 'VALOARE', 'STATUS', 'RECOMANDARE'],
max_rows=10
)
# NEW: Indicatori Lichiditate section
if 'indicatori_lichiditate' in results and not results['indicatori_lichiditate'].empty:
pdf_gen.add_table_section(
"Indicatori de Lichiditate",
results.get('indicatori_lichiditate'),
columns=['INDICATOR', 'VALOARE', 'STATUS', 'RECOMANDARE'],
max_rows=10
)
# NEW: Proiecție Lichiditate
if 'proiectie_lichiditate' in results and not results['proiectie_lichiditate'].empty:
pdf_gen.add_table_section(
"Proiecție Cash Flow 30/60/90 zile",
results.get('proiectie_lichiditate'),
columns=['PERIOADA', 'SOLD_PROIECTAT', 'INCASARI', 'PLATI', 'STATUS'],
max_rows=5
)
# NEW: Recommendations section (top priorities)
if 'recomandari' in results and not results['recomandari'].empty:
pdf_gen.add_recommendations_section(results['recomandari'])
# Alerts
pdf_gen.add_alerts_section({
'vanzari_sub_cost': results.get('vanzari_sub_cost', pd.DataFrame()),
'clienti_marja_mica': results.get('clienti_marja_mica', pd.DataFrame())
})
pdf_gen.add_page_break()
# Monthly chart
if 'vanzari_lunare' in results and not results['vanzari_lunare'].empty:
fig = create_monthly_chart(results['vanzari_lunare'])
pdf_gen.add_chart_image(fig, "Evoluția Vânzărilor și Marjei")
# Client concentration
if 'concentrare_clienti' in results and not results['concentrare_clienti'].empty:
fig = create_client_concentration_chart(results['concentrare_clienti'])
pdf_gen.add_chart_image(fig, "Concentrare Clienți")
pdf_gen.add_page_break()
# NEW: Cash Conversion Cycle chart
if 'ciclu_conversie_cash' in results and not results['ciclu_conversie_cash'].empty:
fig = create_cash_cycle_chart(results['ciclu_conversie_cash'])
pdf_gen.add_chart_image(fig, "Ciclu Conversie Cash (DIO + DSO - DPO)")
# Production vs Resale
if 'productie_vs_revanzare' in results and not results['productie_vs_revanzare'].empty:
fig = create_production_chart(results['productie_vs_revanzare'])
pdf_gen.add_chart_image(fig, "Producție Proprie vs Revânzare")
# Top clients table
pdf_gen.add_table_section(
"Top 15 Clienți după Vânzări",
results.get('marja_per_client'),
columns=['CLIENT', 'VANZARI_FARA_TVA', 'MARJA_BRUTA', 'PROCENT_MARJA'],
max_rows=15
)
pdf_gen.add_page_break()
# Top products
pdf_gen.add_table_section(
"Top 15 Produse după Vânzări",
results.get('top_produse'),
columns=['PRODUS', 'VALOARE_VANZARI', 'MARJA_BRUTA', 'PROCENT_MARJA'],
max_rows=15
)
# Trending clients
pdf_gen.add_table_section(
"Trending Clienți (YoY)",
results.get('trending_clienti'),
columns=['CLIENT', 'VANZARI_12_LUNI', 'VANZARI_AN_ANTERIOR', 'VARIATIE_PROCENT', 'TREND'],
max_rows=15
)
# NEW: Aging Creanțe table
if 'aging_creante' in results and not results['aging_creante'].empty:
pdf_gen.add_page_break()
pdf_gen.add_table_section(
"Aging Creanțe (Vechime Facturi Neîncasate)",
results.get('aging_creante'),
columns=['CLIENT', 'NEAJUNS_SCADENTA', 'ZILE_1_30', 'ZILE_31_60', 'PESTE_90_ZILE', 'TOTAL_SOLD'],
max_rows=15
)
# Stoc lent
if 'stoc_lent' in results and not results['stoc_lent'].empty:
pdf_gen.add_page_break()
pdf_gen.add_table_section(
"Stoc Lent (>90 zile fără mișcare)",
results.get('stoc_lent'),
columns=['PRODUS', 'NUME_GESTIUNE', 'CANTITATE', 'VALOARE', 'ZILE_FARA_MISCARE'],
max_rows=20
)
pdf_gen.save()
# Summary
print("\n" + "="*60)
print(" ✅ RAPOARTE GENERATE CU SUCCES!")
print("="*60)
print(f"\n 📊 Excel: {excel_path}")
print(f" 📄 PDF: {pdf_path}")
print("\n" + "="*60)
return excel_path, pdf_path
def main():
"""Entry point"""
parser = argparse.ArgumentParser(
description='Data Intelligence Report Generator',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Exemple:
python main.py # Raport pentru ultimele 12 luni
python main.py --months 6 # Raport pentru ultimele 6 luni
python main.py --output-dir /tmp # Salvare în alt director
"""
)
parser.add_argument(
'--months', '-m',
type=int,
default=ANALYSIS_MONTHS,
help=f'Numărul de luni pentru analiză (default: {ANALYSIS_MONTHS})'
)
parser.add_argument(
'--output-dir', '-o',
type=Path,
default=OUTPUT_DIR,
help=f'Directorul pentru output (default: {OUTPUT_DIR})'
)
args = parser.parse_args()
# Ensure output directory exists
args.output_dir.mkdir(parents=True, exist_ok=True)
try:
generate_reports(args)
except KeyboardInterrupt:
print("\n\n⚠️ Întrerupt de utilizator.")
sys.exit(1)
except Exception as e:
print(f"\n❌ Eroare: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == '__main__':
main()

2480
queries.py Normal file
View File

File diff suppressed because it is too large Load Diff

581
recommendations.py Normal file
View File

@@ -0,0 +1,581 @@
"""
Recommendations Engine Module
Generates automatic business recommendations based on data analysis
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional
import pandas as pd
@dataclass
class Recommendation:
"""Single business recommendation with context"""
categorie: str # Marja | Clienti | Stoc | Financiar
indicator: str # Indicator name
valoare: str # Current value
status: str # OK | ATENTIE | ALERTA
status_icon: str # OK | ATENTIE | ALERTA (for Excel without emojis)
explicatie: str # Explanation for entrepreneur
recomandare: str # Action suggestion
vezi_detalii: str # Reference to detail sheet
class RecommendationsEngine:
"""Engine that analyzes data and generates recommendations"""
def __init__(self, thresholds: dict):
"""Initialize with threshold configuration"""
self.thresholds = thresholds
self.recommendations: List[Recommendation] = []
def analyze_all(self, results: Dict[str, pd.DataFrame]) -> pd.DataFrame:
"""Run all analyses and return recommendations as DataFrame"""
self.recommendations = []
# Run all analysis modules
self._analyze_margin(results)
self._analyze_clients(results)
self._analyze_stock(results)
self._analyze_financial(results)
self._analyze_general(results)
self._analyze_liquidity(results)
# Convert to DataFrame
if not self.recommendations:
return pd.DataFrame(columns=[
'CATEGORIE', 'INDICATOR', 'VALOARE', 'STATUS',
'EXPLICATIE', 'RECOMANDARE', 'VEZI_DETALII'
])
return pd.DataFrame([
{
'CATEGORIE': r.categorie,
'INDICATOR': r.indicator,
'VALOARE': r.valoare,
'STATUS': r.status_icon,
'EXPLICATIE': r.explicatie,
'RECOMANDARE': r.recomandare,
'VEZI_DETALII': r.vezi_detalii
}
for r in self.recommendations
])
def _add_recommendation(self, categorie: str, indicator: str, valoare: str,
status: str, explicatie: str, recomandare: str,
vezi_detalii: str):
"""Add a recommendation with proper status icon"""
status_icons = {
'OK': 'OK',
'ATENTIE': 'ATENTIE',
'ALERTA': 'ALERTA'
}
self.recommendations.append(Recommendation(
categorie=categorie,
indicator=indicator,
valoare=valoare,
status=status,
status_icon=status_icons.get(status, status),
explicatie=explicatie,
recomandare=recomandare,
vezi_detalii=vezi_detalii
))
def _analyze_margin(self, results: Dict[str, pd.DataFrame]):
"""Analyze margin indicators"""
# 1. Check average margin from sumar_executiv
sumar = results.get('sumar_executiv')
if sumar is not None and not sumar.empty:
marja_row = sumar[sumar['INDICATOR'].str.contains('marj', case=False, na=False)]
if not marja_row.empty:
try:
marja_val = float(str(marja_row['VALOARE'].iloc[0]).strip().replace(',', '.'))
marja_tinta = self.thresholds.get('marja_tinta', 20)
marja_minima = self.thresholds.get('marja_minima', 15)
if marja_val < marja_minima:
self._add_recommendation(
categorie='Marja',
indicator='Marja medie globala',
valoare=f'{marja_val:.1f}%',
status='ALERTA',
explicatie=f'Marja medie de {marja_val:.1f}% este sub pragul minim de {marja_minima}%',
recomandare='Revizuieste urgent preturile sau renegociaza cu furnizorii',
vezi_detalii='Sheet: Marja Per Client'
)
elif marja_val < marja_tinta:
self._add_recommendation(
categorie='Marja',
indicator='Marja medie globala',
valoare=f'{marja_val:.1f}%',
status='ATENTIE',
explicatie=f'Marja medie de {marja_val:.1f}% este sub tinta de {marja_tinta}%',
recomandare='Analizeaza clientii cu marja mica si negociaza preturi',
vezi_detalii='Sheet: Clienti Marja Mica'
)
except (ValueError, IndexError):
pass
# 2. Check sales below cost
vanzari_sub_cost = results.get('vanzari_sub_cost')
if vanzari_sub_cost is not None and not vanzari_sub_cost.empty:
count = len(vanzari_sub_cost)
total_pierdere = abs(vanzari_sub_cost['PIERDERE'].sum()) if 'PIERDERE' in vanzari_sub_cost.columns else 0
self._add_recommendation(
categorie='Marja',
indicator='Vanzari sub cost',
valoare=f'{count} tranzactii',
status='ALERTA',
explicatie=f'{count} produse vandute sub cost cu pierdere totala de {total_pierdere:,.0f} RON',
recomandare='Verifica preturile si opreste vanzarile in pierdere',
vezi_detalii='Sheet: Vanzari Sub Cost'
)
# 3. Check clients with low margin
clienti_marja_mica = results.get('clienti_marja_mica')
if clienti_marja_mica is not None and not clienti_marja_mica.empty:
count = len(clienti_marja_mica)
total_vanzari = clienti_marja_mica['VANZARI_FARA_TVA'].sum() if 'VANZARI_FARA_TVA' in clienti_marja_mica.columns else 0
self._add_recommendation(
categorie='Marja',
indicator='Clienti cu marja mica',
valoare=f'{count} clienti',
status='ATENTIE',
explicatie=f'{count} clienti cu marja sub 15% totalizeaza {total_vanzari:,.0f} RON vanzari',
recomandare='Renegociaza preturile sau ajusteaza conditiile comerciale',
vezi_detalii='Sheet: Clienti Marja Mica'
)
def _analyze_clients(self, results: Dict[str, pd.DataFrame]):
"""Analyze client-related indicators"""
# 1. Check client concentration risk
concentrare_risc = results.get('concentrare_risc')
if concentrare_risc is not None and not concentrare_risc.empty:
# Top 1 client concentration
top1_max = self.thresholds.get('concentrare_top1_max', 25)
top5_max = self.thresholds.get('concentrare_top5_max', 60)
top1_row = concentrare_risc[concentrare_risc['INDICATOR'].str.contains('Top 1', case=False, na=False)]
if not top1_row.empty:
try:
top1_val = float(top1_row['PROCENT'].iloc[0])
if top1_val > top1_max:
self._add_recommendation(
categorie='Clienti',
indicator='Concentrare Top 1 client',
valoare=f'{top1_val:.1f}%',
status='ALERTA',
explicatie=f'Un singur client reprezinta {top1_val:.1f}% din vanzari (risc major)',
recomandare='Diversifica portofoliul urgent, atrage clienti noi',
vezi_detalii='Sheet: Concentrare Risc'
)
except (ValueError, IndexError):
pass
top5_row = concentrare_risc[concentrare_risc['INDICATOR'].str.contains('Top 5', case=False, na=False)]
if not top5_row.empty:
try:
top5_val = float(top5_row['PROCENT'].iloc[0])
if top5_val > top5_max:
self._add_recommendation(
categorie='Clienti',
indicator='Concentrare Top 5 clienti',
valoare=f'{top5_val:.1f}%',
status='ATENTIE',
explicatie=f'Doar 5 clienti genereaza {top5_val:.1f}% din vanzari',
recomandare='Reduce dependenta de clientii mari prin diversificare',
vezi_detalii='Sheet: Concentrare Risc'
)
except (ValueError, IndexError):
pass
# 2. Check from concentrare_clienti if concentrare_risc not available
concentrare_clienti = results.get('concentrare_clienti')
if concentrare_clienti is not None and not concentrare_clienti.empty:
top5_max = self.thresholds.get('concentrare_top5_max', 60)
if len(concentrare_clienti) >= 5 and 'PROCENT_CUMULAT' in concentrare_clienti.columns:
top5_pct = concentrare_clienti['PROCENT_CUMULAT'].iloc[4] if len(concentrare_clienti) > 4 else 0
if top5_pct > top5_max and not any(r.indicator == 'Concentrare Top 5 clienti' for r in self.recommendations):
self._add_recommendation(
categorie='Clienti',
indicator='Concentrare Top 5 clienti',
valoare=f'{top5_pct:.1f}%',
status='ATENTIE',
explicatie=f'{top5_pct:.1f}% din vanzari vin de la doar 5 clienti',
recomandare='Diversifica portofoliul, atrage clienti noi',
vezi_detalii='Sheet: Concentrare Clienti'
)
# 3. Check portfolio health
portofoliu = results.get('portofoliu_clienti')
if portofoliu is not None and not portofoliu.empty:
clienti_pierduti = portofoliu[portofoliu['INDICATOR'].str.contains('pierdut', case=False, na=False)]
if not clienti_pierduti.empty:
try:
pierduti_val = int(clienti_pierduti['VALOARE'].iloc[0])
if pierduti_val > 5:
self._add_recommendation(
categorie='Clienti',
indicator='Clienti pierduti',
valoare=f'{pierduti_val} clienti',
status='ATENTIE',
explicatie=f'{pierduti_val} clienti nu au mai cumparat de peste 6 luni',
recomandare='Contacteaza clientii inactivi pentru reactivare',
vezi_detalii='Sheet: Portofoliu Clienti'
)
except (ValueError, IndexError):
pass
# 4. Check trending clients for negative trends
trending = results.get('trending_clienti')
if trending is not None and not trending.empty and 'TREND' in trending.columns:
pierduti = trending[trending['TREND'] == 'PIERDUT']
scadere = trending[trending['TREND'] == 'SCADERE']
if len(pierduti) > 3:
self._add_recommendation(
categorie='Clienti',
indicator='Clienti pierduti YoY',
valoare=f'{len(pierduti)} clienti',
status='ALERTA',
explicatie=f'{len(pierduti)} clienti activi anul trecut nu au mai cumparat',
recomandare='Investigheaza motivele pierderii si incearca recuperarea',
vezi_detalii='Sheet: Trending Clienti'
)
if len(scadere) > 5:
self._add_recommendation(
categorie='Clienti',
indicator='Clienti in scadere',
valoare=f'{len(scadere)} clienti',
status='ATENTIE',
explicatie=f'{len(scadere)} clienti au scazut achizitiile cu peste 20%',
recomandare='Contacteaza clientii pentru a intelege motivele',
vezi_detalii='Sheet: Trending Clienti'
)
def _analyze_stock(self, results: Dict[str, pd.DataFrame]):
"""Analyze stock-related indicators"""
# 1. Check slow moving stock
stoc_lent = results.get('stoc_lent')
if stoc_lent is not None and not stoc_lent.empty:
stoc_zile_max = self.thresholds.get('stoc_zile_max', 90)
count = len(stoc_lent)
total_valoare = stoc_lent['VALOARE'].sum() if 'VALOARE' in stoc_lent.columns else 0
if count > 10 or total_valoare > 50000:
self._add_recommendation(
categorie='Stoc',
indicator='Stoc lent',
valoare=f'{total_valoare:,.0f} RON',
status='ATENTIE',
explicatie=f'{count} produse fara miscare de peste {stoc_zile_max} zile, valoare {total_valoare:,.0f} RON',
recomandare='Lichideaza stocul lent prin promotii sau retururi la furnizori',
vezi_detalii='Sheet: Stoc Lent'
)
# 2. Check stock rotation
rotatie = results.get('rotatie_stocuri')
if rotatie is not None and not rotatie.empty and 'ZILE_STOC' in rotatie.columns:
stoc_foarte_lent = rotatie[rotatie['ZILE_STOC'] > 365]
if len(stoc_foarte_lent) > 5:
total_val = stoc_foarte_lent['VALOARE_STOC'].sum() if 'VALOARE_STOC' in stoc_foarte_lent.columns else 0
self._add_recommendation(
categorie='Stoc',
indicator='Rotatie foarte lenta',
valoare=f'{len(stoc_foarte_lent)} produse',
status='ALERTA',
explicatie=f'{len(stoc_foarte_lent)} produse cu stoc pentru peste 1 an, valoare {total_val:,.0f} RON',
recomandare='Revizuieste politica de aprovizionare si lichideaza surplusul',
vezi_detalii='Sheet: Rotatie Stocuri'
)
# 3. Check cash conversion cycle
ciclu = results.get('ciclu_conversie_cash')
if ciclu is not None and not ciclu.empty:
dio_row = ciclu[ciclu['INDICATOR'].str.contains('stoc|DIO', case=False, na=False)]
if not dio_row.empty:
try:
dio_val = float(dio_row['ZILE'].iloc[0])
if dio_val > 90:
self._add_recommendation(
categorie='Stoc',
indicator='Zile stoc (DIO)',
valoare=f'{dio_val:.0f} zile',
status='ATENTIE',
explicatie=f'Stocul sta in medie {dio_val:.0f} zile inainte de vanzare',
recomandare='Optimizeaza aprovizionarea pentru rotatie mai rapida',
vezi_detalii='Sheet: Ciclu Conversie Cash'
)
except (ValueError, IndexError):
pass
def _analyze_financial(self, results: Dict[str, pd.DataFrame]):
"""Analyze financial indicators"""
dso_target = self.thresholds.get('dso_target', 45)
dso_alert = self.thresholds.get('dso_alert', 60)
restante_90_procent = self.thresholds.get('restante_90_procent', 10)
# 1. Check DSO (Days Sales Outstanding)
dso_dpo = results.get('dso_dpo')
if dso_dpo is not None and not dso_dpo.empty:
dso_row = dso_dpo[dso_dpo['INDICATOR'].str.contains('DSO|incasare', case=False, na=False)]
if not dso_row.empty:
try:
dso_val = float(dso_row['ZILE'].iloc[0])
if dso_val > dso_alert:
self._add_recommendation(
categorie='Financiar',
indicator='DSO (Zile incasare)',
valoare=f'{dso_val:.0f} zile',
status='ALERTA',
explicatie=f'Incasezi in medie in {dso_val:.0f} zile, cu mult peste tinta de {dso_target} zile',
recomandare='Implementeaza urmarire stricta a incasarilor si penalitati',
vezi_detalii='Sheet: DSO DPO'
)
elif dso_val > dso_target:
self._add_recommendation(
categorie='Financiar',
indicator='DSO (Zile incasare)',
valoare=f'{dso_val:.0f} zile',
status='ATENTIE',
explicatie=f'Incasezi in medie in {dso_val:.0f} zile, peste tinta de {dso_target} zile',
recomandare='Ofera discount pentru plata rapida (ex: 2% la 10 zile)',
vezi_detalii='Sheet: DSO DPO'
)
except (ValueError, IndexError):
pass
# 2. Check aging receivables
aging = results.get('aging_creante')
if aging is not None and not aging.empty:
if 'PESTE_90_ZILE' in aging.columns and 'TOTAL_SOLD' in aging.columns:
total_creante = aging['TOTAL_SOLD'].sum()
peste_90 = aging['PESTE_90_ZILE'].sum()
if total_creante > 0:
procent_90 = (peste_90 / total_creante) * 100
if procent_90 > restante_90_procent:
self._add_recommendation(
categorie='Financiar',
indicator='Creante >90 zile',
valoare=f'{peste_90:,.0f} RON ({procent_90:.1f}%)',
status='ALERTA',
explicatie=f'{procent_90:.1f}% din creante sunt restante peste 90 zile',
recomandare='Initiaza proceduri de recuperare pentru restantele vechi',
vezi_detalii='Sheet: Aging Creante'
)
# 3. Check overdue invoices
facturi_restante = results.get('facturi_restante')
if facturi_restante is not None and not facturi_restante.empty:
count = len(facturi_restante)
total_restant = facturi_restante['SUMA_RESTANTA'].sum() if 'SUMA_RESTANTA' in facturi_restante.columns else 0
if count > 0:
zile_max = facturi_restante['ZILE_INTARZIERE'].max() if 'ZILE_INTARZIERE' in facturi_restante.columns else 0
status = 'ALERTA' if zile_max > 90 or total_restant > 100000 else 'ATENTIE'
self._add_recommendation(
categorie='Financiar',
indicator='Facturi restante',
valoare=f'{total_restant:,.0f} RON',
status=status,
explicatie=f'{count} facturi restante, cea mai veche de {zile_max:.0f} zile',
recomandare='Prioritizeaza incasarea facturilor cele mai vechi',
vezi_detalii='Sheet: Facturi Restante'
)
# 4. Check cash conversion cycle
ciclu = results.get('ciclu_conversie_cash')
if ciclu is not None and not ciclu.empty:
ciclu_row = ciclu[ciclu['INDICATOR'].str.contains('ciclu|total|CCC', case=False, na=False)]
if not ciclu_row.empty:
try:
ciclu_val = float(ciclu_row['ZILE'].iloc[0])
if ciclu_val > 90:
self._add_recommendation(
categorie='Financiar',
indicator='Ciclu conversie cash',
valoare=f'{ciclu_val:.0f} zile',
status='ALERTA',
explicatie=f'Dureaza {ciclu_val:.0f} zile de la plata furnizor pana la incasare',
recomandare='Negociaza termene mai lungi cu furnizorii si mai scurte cu clientii',
vezi_detalii='Sheet: Ciclu Conversie Cash'
)
elif ciclu_val > 60:
self._add_recommendation(
categorie='Financiar',
indicator='Ciclu conversie cash',
valoare=f'{ciclu_val:.0f} zile',
status='ATENTIE',
explicatie=f'Ciclul de conversie cash este de {ciclu_val:.0f} zile',
recomandare='Optimizeaza stocurile si accelereaza incasarile',
vezi_detalii='Sheet: Ciclu Conversie Cash'
)
except (ValueError, IndexError):
pass
# 5. Check client balances for large debtors
solduri_clienti = results.get('solduri_clienti')
if solduri_clienti is not None and not solduri_clienti.empty:
total_creante = solduri_clienti['SOLD_CURENT'].sum() if 'SOLD_CURENT' in solduri_clienti.columns else 0
if total_creante > 0 and len(solduri_clienti) > 0:
top_client = solduri_clienti.iloc[0]
top_sold = top_client.get('SOLD_CURENT', 0)
top_name = top_client.get('CLIENT', 'N/A')
procent_top = (top_sold / total_creante) * 100 if total_creante > 0 else 0
if procent_top > 30:
self._add_recommendation(
categorie='Financiar',
indicator='Concentrare creante',
valoare=f'{procent_top:.1f}% la un client',
status='ATENTIE',
explicatie=f'{top_name[:30]} datoreaza {top_sold:,.0f} RON ({procent_top:.1f}% din total)',
recomandare='Monitorizeaza indeaproape si seteaza limite de credit',
vezi_detalii='Sheet: Solduri Clienti'
)
def _analyze_general(self, results: Dict[str, pd.DataFrame]):
"""Analyze general business indicators (grad indatorare, ROA, autonomie financiara)"""
indicatori_generali = results.get('indicatori_generali')
if indicatori_generali is None or indicatori_generali.empty:
return
# Iterate through each indicator
for _, row in indicatori_generali.iterrows():
try:
indicator = row.get('INDICATOR', '')
valoare = row.get('VALOARE', 0)
status = row.get('STATUS', 'OK')
recomandare_db = row.get('RECOMANDARE', '')
# Skip if no alert needed
if status == 'OK':
continue
# Map indicator to category
if isinstance(valoare, (int, float)):
valoare_str = f'{valoare:.2f}'
else:
valoare_str = str(valoare)
self._add_recommendation(
categorie='General',
indicator=indicator,
valoare=valoare_str,
status=status,
explicatie=row.get('INTERPRETARE', ''),
recomandare=recomandare_db,
vezi_detalii='Sheet: Indicatori Generali'
)
except (ValueError, IndexError, KeyError):
pass
def _analyze_liquidity(self, results: Dict[str, pd.DataFrame]):
"""Analyze liquidity indicators"""
# 1. Check basic liquidity indicators
indicatori_lichiditate = results.get('indicatori_lichiditate')
if indicatori_lichiditate is not None and not indicatori_lichiditate.empty:
for _, row in indicatori_lichiditate.iterrows():
try:
indicator = row.get('INDICATOR', '')
valoare = row.get('VALOARE', 0)
status = row.get('STATUS', 'OK')
recomandare_db = row.get('RECOMANDARE', '')
if status == 'OK':
continue
if isinstance(valoare, (int, float)):
valoare_str = f'{valoare:.2f}'
else:
valoare_str = str(valoare)
self._add_recommendation(
categorie='Lichiditate',
indicator=indicator,
valoare=valoare_str,
status=status,
explicatie=row.get('INTERPRETARE', ''),
recomandare=recomandare_db,
vezi_detalii='Sheet: Indicatori Lichiditate'
)
except (ValueError, IndexError, KeyError):
pass
# 2. Check coverage ratio
grad_acoperire = results.get('grad_acoperire_datorii')
if grad_acoperire is not None and not grad_acoperire.empty:
# Look for the coverage ratio row
acoperire_row = grad_acoperire[grad_acoperire['INDICATOR'].str.contains('GRAD ACOPERIRE', case=False, na=False)]
if not acoperire_row.empty:
try:
acoperire_val = float(acoperire_row['VALOARE'].iloc[0])
acoperire_status = acoperire_row['ACOPERIRE'].iloc[0]
if acoperire_status == 'DEFICIT':
# Look for financing need
necesar_row = grad_acoperire[grad_acoperire['INDICATOR'].str.contains('NECESAR FINANTARE', case=False, na=False)]
necesar_val = float(necesar_row['VALOARE'].iloc[0]) if not necesar_row.empty else 0
self._add_recommendation(
categorie='Lichiditate',
indicator='Grad acoperire datorii 30 zile',
valoare=f'{acoperire_val:.2f}',
status='ALERTA',
explicatie=f'Cash + incasari asteptate nu acopera datoriile scadente. Deficit: {necesar_val:,.0f} RON',
recomandare='Urgentati incasarile sau obtineti finantare pe termen scurt',
vezi_detalii='Sheet: Grad Acoperire Datorii'
)
elif acoperire_status == 'ATENTIE':
self._add_recommendation(
categorie='Lichiditate',
indicator='Grad acoperire datorii 30 zile',
valoare=f'{acoperire_val:.2f}',
status='ATENTIE',
explicatie='Acoperirea datoriilor este la limita',
recomandare='Mentineti rezerve suplimentare pentru imprevaute',
vezi_detalii='Sheet: Grad Acoperire Datorii'
)
except (ValueError, IndexError, KeyError):
pass
# 3. Check cash projection
proiectie = results.get('proiectie_lichiditate')
if proiectie is not None and not proiectie.empty:
for _, row in proiectie.iterrows():
try:
perioada = row.get('PERIOADA', '')
sold = row.get('SOLD_PROIECTAT', 0)
status = row.get('STATUS', 'OK')
if status == 'ALERTA' and 'Proiectie' in perioada:
self._add_recommendation(
categorie='Lichiditate',
indicator=f'Proiectie cash: {perioada}',
valoare=f'{sold:,.0f} RON',
status='ALERTA',
explicatie=f'Sold cash proiectat negativ la {perioada}',
recomandare='Planificati finantare sau accelerati incasarile pentru a evita criza de lichiditate',
vezi_detalii='Sheet: Proiectie Lichiditate'
)
break # Only report first negative projection
except (ValueError, IndexError, KeyError):
pass

773
report_generator.py Normal file
View File

@@ -0,0 +1,773 @@
"""
Report Generator Module
Generates Excel and PDF reports from query results
"""
import pandas as pd
from datetime import datetime
from pathlib import Path
import matplotlib
matplotlib.use('Agg') # Non-interactive backend
import matplotlib.pyplot as plt
import matplotlib.ticker as ticker
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
from openpyxl.utils.dataframe import dataframe_to_rows
from openpyxl.chart import BarChart, LineChart, PieChart, Reference
from openpyxl.utils import get_column_letter
from reportlab.lib import colors
from reportlab.lib.pagesizes import A4, landscape
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import cm, mm
from reportlab.platypus import (
SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle,
PageBreak, Image, KeepTogether
)
from reportlab.lib.enums import TA_CENTER, TA_LEFT, TA_RIGHT
import io
class ExcelReportGenerator:
"""Generate Excel reports with multiple sheets and formatting"""
def __init__(self, output_path: Path):
self.output_path = output_path
self.wb = Workbook()
# Remove default sheet
self.wb.remove(self.wb.active)
# Define styles
self.header_font = Font(bold=True, color='FFFFFF', size=11)
self.header_fill = PatternFill(start_color='366092', end_color='366092', fill_type='solid')
self.alert_fill = PatternFill(start_color='FF6B6B', end_color='FF6B6B', fill_type='solid')
self.warning_fill = PatternFill(start_color='FFE66D', end_color='FFE66D', fill_type='solid')
self.good_fill = PatternFill(start_color='4ECDC4', end_color='4ECDC4', fill_type='solid')
self.border = Border(
left=Side(style='thin'),
right=Side(style='thin'),
top=Side(style='thin'),
bottom=Side(style='thin')
)
def add_sheet(self, name: str, df: pd.DataFrame, title: str = None, description: str = None, legend: dict = None):
"""Add a formatted sheet to the workbook with optional legend"""
# Truncate sheet name to 31 chars (Excel limit)
sheet_name = name[:31]
ws = self.wb.create_sheet(title=sheet_name)
start_row = 1
# Add title if provided
if title:
ws.cell(row=start_row, column=1, value=title)
ws.cell(row=start_row, column=1).font = Font(bold=True, size=14)
start_row += 1
# Add description if provided
if description:
ws.cell(row=start_row, column=1, value=description)
ws.cell(row=start_row, column=1).font = Font(italic=True, size=10, color='666666')
start_row += 1
# Add timestamp
ws.cell(row=start_row, column=1, value=f"Generat: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
ws.cell(row=start_row, column=1).font = Font(size=9, color='999999')
start_row += 1
# Add legend if provided
if legend:
start_row += 1
ws.cell(row=start_row, column=1, value="Explicații calcule:")
ws.cell(row=start_row, column=1).font = Font(bold=True, size=9, color='336699')
start_row += 1
for col_name, explanation in legend.items():
ws.cell(row=start_row, column=1, value=f"• {col_name}: {explanation}")
ws.cell(row=start_row, column=1).font = Font(size=8, color='666666')
start_row += 1
start_row += 1
if df is None or df.empty:
ws.cell(row=start_row, column=1, value="Nu există date pentru această analiză.")
return
# Write headers
for col_idx, col_name in enumerate(df.columns, 1):
cell = ws.cell(row=start_row, column=col_idx, value=col_name)
cell.font = self.header_font
cell.fill = self.header_fill
cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
cell.border = self.border
# Write data
for row_idx, row in enumerate(df.itertuples(index=False), start_row + 1):
for col_idx, value in enumerate(row, 1):
cell = ws.cell(row=row_idx, column=col_idx, value=value)
cell.border = self.border
# Format numbers
if isinstance(value, (int, float)):
cell.number_format = '#,##0.00' if isinstance(value, float) else '#,##0'
cell.alignment = Alignment(horizontal='right')
# Highlight negative values
if isinstance(value, (int, float)) and value < 0:
cell.fill = self.alert_fill
# Highlight low margins
col_name = df.columns[col_idx - 1].lower()
if 'procent' in col_name or 'marja' in col_name:
if isinstance(value, (int, float)):
if value < 10:
cell.fill = self.alert_fill
elif value < 15:
cell.fill = self.warning_fill
elif value > 25:
cell.fill = self.good_fill
# Highlight TREND column for YoY sheets
if col_name == 'trend':
if isinstance(value, str):
if value in ('CRESTERE', 'IMBUNATATIRE', 'DIVERSIFICARE'):
cell.fill = self.good_fill
elif value in ('SCADERE', 'DETERIORARE', 'CONCENTRARE', 'PIERDUT'):
cell.fill = self.alert_fill
elif value == 'ATENTIE':
cell.fill = self.warning_fill
# Highlight STATUS column
if col_name == 'status' or col_name == 'acoperire':
if isinstance(value, str):
if value == 'OK':
cell.fill = self.good_fill
elif value in ('ATENTIE', 'NECESAR'):
cell.fill = self.warning_fill
elif value in ('ALERTA', 'DEFICIT', 'RISC MARE'):
cell.fill = self.alert_fill
# Highlight variatie columns (positive = green, negative = red)
if 'variatie' in col_name:
if isinstance(value, (int, float)):
if value > 0:
cell.fill = self.good_fill
elif value < 0:
cell.fill = self.alert_fill
# Auto-adjust column widths
for col_idx, col_name in enumerate(df.columns, 1):
max_length = len(str(col_name))
for row in df.itertuples(index=False):
cell_value = row[col_idx - 1]
if cell_value:
max_length = max(max_length, len(str(cell_value)))
adjusted_width = min(max_length + 2, 50)
ws.column_dimensions[get_column_letter(col_idx)].width = adjusted_width
# Freeze header row
ws.freeze_panes = ws.cell(row=start_row + 1, column=1)
def add_sheet_with_recommendations(self, name: str, df: pd.DataFrame,
recommendations_df: pd.DataFrame,
title: str = None, description: str = None,
legend: dict = None, top_n_recommendations: int = 5):
"""Adauga sheet formatat cu KPIs si top recomandari dedesubt"""
sheet_name = name[:31]
ws = self.wb.create_sheet(title=sheet_name)
start_row = 1
# Adauga titlu
if title:
ws.cell(row=start_row, column=1, value=title)
ws.cell(row=start_row, column=1).font = Font(bold=True, size=14)
start_row += 1
# Adauga descriere
if description:
ws.cell(row=start_row, column=1, value=description)
ws.cell(row=start_row, column=1).font = Font(italic=True, size=10, color='666666')
start_row += 1
# Adauga timestamp
ws.cell(row=start_row, column=1, value=f"Generat: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
ws.cell(row=start_row, column=1).font = Font(size=9, color='999999')
start_row += 2
# === SECTIUNEA 1: KPIs ===
if df is not None and not df.empty:
# Header
for col_idx, col_name in enumerate(df.columns, 1):
cell = ws.cell(row=start_row, column=col_idx, value=col_name)
cell.font = self.header_font
cell.fill = self.header_fill
cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
cell.border = self.border
# Date
for row_idx, row in enumerate(df.itertuples(index=False), start_row + 1):
for col_idx, value in enumerate(row, 1):
cell = ws.cell(row=row_idx, column=col_idx, value=value)
cell.border = self.border
if isinstance(value, (int, float)):
cell.number_format = '#,##0.00' if isinstance(value, float) else '#,##0'
cell.alignment = Alignment(horizontal='right')
start_row = start_row + len(df) + 3
# === SECTIUNEA 2: TOP RECOMANDARI ===
if recommendations_df is not None and not recommendations_df.empty:
ws.cell(row=start_row, column=1, value="Top Recomandari Prioritare")
ws.cell(row=start_row, column=1).font = Font(bold=True, size=12, color='366092')
start_row += 1
# Sorteaza dupa prioritate (ALERTA primul, apoi ATENTIE, apoi OK)
df_sorted = recommendations_df.copy()
status_order = {'ALERTA': 0, 'ATENTIE': 1, 'OK': 2}
df_sorted['_order'] = df_sorted['STATUS'].map(status_order).fillna(3)
df_sorted = df_sorted.sort_values('_order').head(top_n_recommendations)
df_sorted = df_sorted.drop(columns=['_order'])
# Coloane de afisat
display_cols = ['STATUS', 'CATEGORIE', 'INDICATOR', 'VALOARE', 'RECOMANDARE']
display_cols = [c for c in display_cols if c in df_sorted.columns]
# Header cu background mov
rec_header_fill = PatternFill(start_color='8E44AD', end_color='8E44AD', fill_type='solid')
for col_idx, col_name in enumerate(display_cols, 1):
cell = ws.cell(row=start_row, column=col_idx, value=col_name)
cell.font = self.header_font
cell.fill = rec_header_fill
cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True)
cell.border = self.border
# Randuri cu colorare dupa status
for row_idx, (_, row) in enumerate(df_sorted.iterrows(), start_row + 1):
status = row.get('STATUS', 'OK')
for col_idx, col_name in enumerate(display_cols, 1):
value = row.get(col_name, '')
cell = ws.cell(row=row_idx, column=col_idx, value=value)
cell.border = self.border
cell.alignment = Alignment(wrap_text=True)
# Colorare conditionala
if status == 'ALERTA':
cell.fill = PatternFill(start_color='FADBD8', end_color='FADBD8', fill_type='solid')
elif status == 'ATENTIE':
cell.fill = PatternFill(start_color='FCF3CF', end_color='FCF3CF', fill_type='solid')
else:
cell.fill = PatternFill(start_color='D5F5E3', end_color='D5F5E3', fill_type='solid')
# Auto-adjust latime coloane
for col_idx in range(1, 8):
ws.column_dimensions[get_column_letter(col_idx)].width = 22
ws.freeze_panes = ws.cell(row=5, column=1)
def save(self):
"""Save the workbook"""
self.wb.save(self.output_path)
print(f"✓ Excel salvat: {self.output_path}")
class PDFReportGenerator:
"""Generate PDF executive summary with charts"""
def __init__(self, output_path: Path, company_name: str = "Data Intelligence Report"):
self.output_path = output_path
self.company_name = company_name
self.elements = []
self.styles = getSampleStyleSheet()
# Custom styles
self.styles.add(ParagraphStyle(
name='CustomTitle',
parent=self.styles['Title'],
fontSize=24,
spaceAfter=30,
alignment=TA_CENTER
))
self.styles.add(ParagraphStyle(
name='SectionHeader',
parent=self.styles['Heading1'],
fontSize=14,
spaceBefore=20,
spaceAfter=10,
textColor=colors.HexColor('#366092')
))
self.styles.add(ParagraphStyle(
name='AlertHeader',
parent=self.styles['Heading2'],
fontSize=12,
textColor=colors.red,
spaceBefore=15,
spaceAfter=8
))
self.styles.add(ParagraphStyle(
name='SmallText',
parent=self.styles['Normal'],
fontSize=8,
textColor=colors.gray
))
def add_title_page(self, report_date: datetime = None):
"""Add title page"""
if report_date is None:
report_date = datetime.now()
self.elements.append(Spacer(1, 3*cm))
self.elements.append(Paragraph(self.company_name, self.styles['CustomTitle']))
self.elements.append(Spacer(1, 1*cm))
self.elements.append(Paragraph(
f"Raport generat: {report_date.strftime('%d %B %Y, %H:%M')}",
self.styles['Normal']
))
self.elements.append(Paragraph(
"Perioada analizată: Ultimele 12 luni",
self.styles['Normal']
))
self.elements.append(PageBreak())
def add_kpi_section(self, kpi_df: pd.DataFrame):
"""Add KPI summary section"""
self.elements.append(Paragraph("📊 Sumar Executiv - KPIs", self.styles['SectionHeader']))
if kpi_df is not None and not kpi_df.empty:
data = [['Indicator', 'Valoare', 'UM']]
for _, row in kpi_df.iterrows():
data.append([
str(row.get('INDICATOR', '')),
str(row.get('VALOARE', '')),
str(row.get('UM', ''))
])
table = Table(data, colWidths=[8*cm, 4*cm, 2*cm])
table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#366092')),
('TEXTCOLOR', (0, 0), (-1, 0), colors.white),
('ALIGN', (0, 0), (-1, -1), 'LEFT'),
('ALIGN', (1, 1), (1, -1), 'RIGHT'),
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
('FONTSIZE', (0, 0), (-1, 0), 10),
('BOTTOMPADDING', (0, 0), (-1, 0), 12),
('GRID', (0, 0), (-1, -1), 0.5, colors.gray),
('ROWBACKGROUNDS', (0, 1), (-1, -1), [colors.white, colors.HexColor('#f0f0f0')])
]))
self.elements.append(table)
self.elements.append(Spacer(1, 0.5*cm))
def add_alerts_section(self, alerts_data: dict):
"""Add critical alerts section"""
self.elements.append(Paragraph("🚨 Alerte Critice", self.styles['SectionHeader']))
# Vânzări sub cost
if 'vanzari_sub_cost' in alerts_data and not alerts_data['vanzari_sub_cost'].empty:
df = alerts_data['vanzari_sub_cost']
count = len(df)
total_loss = df['PIERDERE'].sum() if 'PIERDERE' in df.columns else 0
self.elements.append(Paragraph(
f"⛔ VÂNZĂRI SUB COST: {count} tranzacții cu pierdere totală de {abs(total_loss):,.2f} RON",
self.styles['AlertHeader']
))
# Show top 5
top5 = df.head(5)
if not top5.empty:
cols_to_show = ['FACTURA', 'CLIENT', 'PRODUS', 'PIERDERE']
cols_to_show = [c for c in cols_to_show if c in top5.columns]
if cols_to_show:
data = [cols_to_show] + top5[cols_to_show].values.tolist()
table = Table(data, colWidths=[3*cm, 4*cm, 5*cm, 2*cm])
table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#c0392b')),
('TEXTCOLOR', (0, 0), (-1, 0), colors.white),
('FONTSIZE', (0, 0), (-1, -1), 8),
('GRID', (0, 0), (-1, -1), 0.5, colors.gray),
]))
self.elements.append(table)
# Clienți cu marjă mică
if 'clienti_marja_mica' in alerts_data and not alerts_data['clienti_marja_mica'].empty:
df = alerts_data['clienti_marja_mica']
count = len(df)
self.elements.append(Spacer(1, 0.3*cm))
self.elements.append(Paragraph(
f"⚠️ CLIENȚI CU MARJĂ MICĂ (<15%): {count} clienți necesită renegociere",
self.styles['AlertHeader']
))
top5 = df.head(5)
if not top5.empty:
cols_to_show = ['CLIENT', 'VANZARI_FARA_TVA', 'PROCENT_MARJA']
cols_to_show = [c for c in cols_to_show if c in top5.columns]
if cols_to_show:
data = [cols_to_show]
for _, row in top5.iterrows():
data.append([
str(row.get('CLIENT', ''))[:30],
f"{row.get('VANZARI_FARA_TVA', 0):,.0f}",
f"{row.get('PROCENT_MARJA', 0):.1f}%"
])
table = Table(data, colWidths=[6*cm, 3*cm, 2*cm])
table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#e67e22')),
('TEXTCOLOR', (0, 0), (-1, 0), colors.white),
('FONTSIZE', (0, 0), (-1, -1), 8),
('GRID', (0, 0), (-1, -1), 0.5, colors.gray),
]))
self.elements.append(table)
self.elements.append(Spacer(1, 0.5*cm))
def add_chart_image(self, fig, title: str):
"""Add a matplotlib figure as image"""
self.elements.append(Paragraph(title, self.styles['SectionHeader']))
# Save figure to buffer
buf = io.BytesIO()
fig.savefig(buf, format='png', dpi=150, bbox_inches='tight')
buf.seek(0)
# Add to PDF
img = Image(buf, width=16*cm, height=10*cm)
self.elements.append(img)
self.elements.append(Spacer(1, 0.5*cm))
plt.close(fig)
def add_table_section(self, title: str, df: pd.DataFrame, columns: list = None, max_rows: int = 15):
"""Add a data table section"""
self.elements.append(Paragraph(title, self.styles['SectionHeader']))
if df is None or df.empty:
self.elements.append(Paragraph("Nu există date.", self.styles['Normal']))
return
# Select columns
if columns:
cols = [c for c in columns if c in df.columns]
else:
cols = list(df.columns)[:6] # Max 6 columns for PDF
if not cols:
return
# Prepare data
data = [cols]
for _, row in df.head(max_rows).iterrows():
row_data = []
for col in cols:
val = row.get(col, '')
if isinstance(val, float):
row_data.append(f"{val:,.2f}")
elif isinstance(val, int):
row_data.append(f"{val:,}")
else:
row_data.append(str(val)[:25]) # Truncate long strings
data.append(row_data)
# Calculate column widths
n_cols = len(cols)
col_width = 16*cm / n_cols
table = Table(data, colWidths=[col_width] * n_cols)
table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#366092')),
('TEXTCOLOR', (0, 0), (-1, 0), colors.white),
('ALIGN', (0, 0), (-1, -1), 'LEFT'),
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
('FONTSIZE', (0, 0), (-1, -1), 7),
('BOTTOMPADDING', (0, 0), (-1, 0), 8),
('GRID', (0, 0), (-1, -1), 0.5, colors.gray),
('ROWBACKGROUNDS', (0, 1), (-1, -1), [colors.white, colors.HexColor('#f5f5f5')])
]))
self.elements.append(table)
if len(df) > max_rows:
self.elements.append(Paragraph(
f"... și încă {len(df) - max_rows} înregistrări (vezi Excel pentru lista completă)",
self.styles['SmallText']
))
self.elements.append(Spacer(1, 0.5*cm))
def add_page_break(self):
"""Add page break"""
self.elements.append(PageBreak())
def add_recommendations_section(self, recommendations_df: pd.DataFrame):
"""Add recommendations section with status colors"""
self.elements.append(Paragraph("Recomandari Cheie", self.styles['SectionHeader']))
if recommendations_df is None or recommendations_df.empty:
self.elements.append(Paragraph("Nu au fost generate recomandari.", self.styles['Normal']))
return
# Show top 7 most important recommendations (ALERTA first, then ATENTIE)
df_sorted = recommendations_df.copy()
status_order = {'ALERTA': 0, 'ATENTIE': 1, 'OK': 2}
df_sorted['_order'] = df_sorted['STATUS'].map(status_order)
df_sorted = df_sorted.sort_values('_order').head(7)
for _, row in df_sorted.iterrows():
status = row.get('STATUS', 'OK')
indicator = row.get('INDICATOR', '')
valoare = row.get('VALOARE', '')
explicatie = row.get('EXPLICATIE', '')
recomandare = row.get('RECOMANDARE', '')
# Color based on status
if status == 'ALERTA':
status_color = colors.HexColor('#c0392b')
bg_color = colors.HexColor('#fadbd8')
elif status == 'ATENTIE':
status_color = colors.HexColor('#d68910')
bg_color = colors.HexColor('#fef9e7')
else:
status_color = colors.HexColor('#27ae60')
bg_color = colors.HexColor('#d5f5e3')
# Create a small table for each recommendation
data = [
[f"[{status}] {indicator}: {valoare}"],
[explicatie],
[f"Actiune: {recomandare}"]
]
table = Table(data, colWidths=[16*cm])
table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), status_color),
('TEXTCOLOR', (0, 0), (-1, 0), colors.white),
('BACKGROUND', (0, 1), (-1, -1), bg_color),
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
('FONTSIZE', (0, 0), (-1, -1), 8),
('BOTTOMPADDING', (0, 0), (-1, -1), 6),
('TOPPADDING', (0, 0), (-1, -1), 4),
('LEFTPADDING', (0, 0), (-1, -1), 8),
]))
self.elements.append(table)
self.elements.append(Spacer(1, 0.2*cm))
self.elements.append(Spacer(1, 0.3*cm))
def save(self):
"""Generate and save PDF"""
doc = SimpleDocTemplate(
str(self.output_path),
pagesize=A4,
rightMargin=2*cm,
leftMargin=2*cm,
topMargin=2*cm,
bottomMargin=2*cm
)
doc.build(self.elements)
print(f"✓ PDF salvat: {self.output_path}")
def create_monthly_chart(df: pd.DataFrame) -> plt.Figure:
"""Create monthly sales and margin chart"""
if df is None or df.empty:
fig, ax = plt.subplots(figsize=(12, 6))
ax.text(0.5, 0.5, 'Nu există date', ha='center', va='center')
return fig
fig, ax1 = plt.subplots(figsize=(12, 6))
x = range(len(df))
# Bar chart for sales
bars = ax1.bar(x, df['VANZARI_FARA_TVA'], color='#366092', alpha=0.7, label='Vânzări')
ax1.set_xlabel('Luna')
ax1.set_ylabel('Vânzări (RON)', color='#366092')
ax1.tick_params(axis='y', labelcolor='#366092')
ax1.yaxis.set_major_formatter(ticker.FuncFormatter(lambda x, p: f'{x/1000:,.0f}k'))
# Line chart for margin
ax2 = ax1.twinx()
line = ax2.plot(x, df['MARJA_BRUTA'], color='#e74c3c', linewidth=2, marker='o', label='Marja')
ax2.set_ylabel('Marja (RON)', color='#e74c3c')
ax2.tick_params(axis='y', labelcolor='#e74c3c')
ax2.yaxis.set_major_formatter(ticker.FuncFormatter(lambda x, p: f'{x/1000:,.0f}k'))
# X-axis labels
ax1.set_xticks(x)
ax1.set_xticklabels(df['LUNA'], rotation=45, ha='right')
# Legend
lines1, labels1 = ax1.get_legend_handles_labels()
lines2, labels2 = ax2.get_legend_handles_labels()
ax1.legend(lines1 + lines2, labels1 + labels2, loc='upper left')
plt.title('Evoluția Vânzărilor și Marjei Lunare')
plt.tight_layout()
return fig
def create_client_concentration_chart(df: pd.DataFrame) -> plt.Figure:
"""Create client concentration pie chart"""
if df is None or df.empty:
fig, ax = plt.subplots(figsize=(10, 8))
ax.text(0.5, 0.5, 'Nu există date', ha='center', va='center')
return fig
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 6))
# Pie chart - Top 10 vs Others
top10 = df.head(10)
others_pct = 100 - top10['PROCENT_CUMULAT'].iloc[-1] if len(top10) >= 10 else 0
sizes = list(top10['PROCENT_DIN_TOTAL'])
if others_pct > 0:
sizes.append(others_pct)
labels = list(top10['CLIENT'].str[:20]) # Truncate names
if others_pct > 0:
labels.append('Alții')
colors_list = plt.cm.Set3(range(len(sizes)))
ax1.pie(sizes, labels=None, colors=colors_list, autopct='%1.1f%%', startangle=90)
ax1.set_title('Concentrare Top 10 Clienți')
ax1.legend(labels, loc='center left', bbox_to_anchor=(1, 0.5), fontsize=8)
# Bar chart - Pareto
ax2.bar(range(len(top10)), top10['VANZARI'], color='#366092', alpha=0.7)
ax2_twin = ax2.twinx()
ax2_twin.plot(range(len(top10)), top10['PROCENT_CUMULAT'], 'r-o', linewidth=2)
ax2_twin.axhline(y=80, color='green', linestyle='--', alpha=0.5, label='80%')
ax2.set_xticks(range(len(top10)))
ax2.set_xticklabels([c[:15] for c in top10['CLIENT']], rotation=45, ha='right', fontsize=8)
ax2.set_ylabel('Vânzări (RON)')
ax2_twin.set_ylabel('% Cumulat')
ax2.set_title('Analiză Pareto Clienți')
plt.tight_layout()
return fig
def create_production_chart(df: pd.DataFrame) -> plt.Figure:
"""Create production vs resale comparison chart"""
if df is None or df.empty:
fig, ax = plt.subplots(figsize=(10, 6))
ax.text(0.5, 0.5, 'Nu există date', ha='center', va='center')
return fig
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))
# Bar chart - Sales by type
x = range(len(df))
ax1.bar(x, df['VANZARI_FARA_TVA'], color=['#366092', '#e74c3c', '#2ecc71'][:len(df)])
ax1.set_xticks(x)
ax1.set_xticklabels(df['TIP_PRODUS'], rotation=15)
ax1.set_ylabel('Vânzări (RON)')
ax1.set_title('Vânzări per Tip Produs')
ax1.yaxis.set_major_formatter(ticker.FuncFormatter(lambda x, p: f'{x/1000:,.0f}k'))
# Bar chart - Margin %
colors = ['#2ecc71' if m > 20 else '#e67e22' if m > 15 else '#e74c3c' for m in df['PROCENT_MARJA']]
ax2.bar(x, df['PROCENT_MARJA'], color=colors)
ax2.set_xticks(x)
ax2.set_xticklabels(df['TIP_PRODUS'], rotation=15)
ax2.set_ylabel('Marjă (%)')
ax2.set_title('Marjă per Tip Produs')
ax2.axhline(y=15, color='red', linestyle='--', alpha=0.5, label='Prag minim 15%')
ax2.legend()
plt.tight_layout()
return fig
def create_cash_cycle_chart(df: pd.DataFrame) -> plt.Figure:
"""Create cash conversion cycle visualization"""
if df is None or df.empty:
fig, ax = plt.subplots(figsize=(10, 6))
ax.text(0.5, 0.5, 'Nu exista date', ha='center', va='center')
return fig
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 6))
# Extract values
indicators = df['INDICATOR'].tolist() if 'INDICATOR' in df.columns else []
zile = df['ZILE'].tolist() if 'ZILE' in df.columns else []
if not indicators or not zile:
ax1.text(0.5, 0.5, 'Date incomplete', ha='center', va='center')
ax2.text(0.5, 0.5, 'Date incomplete', ha='center', va='center')
return fig
# Colors for each component
colors_map = {
'DIO': '#3498db', # Blue for inventory
'DSO': '#e74c3c', # Red for receivables
'DPO': '#2ecc71', # Green for payables
'CCC': '#9b59b6' # Purple for total cycle
}
bar_colors = []
for ind in indicators:
for key, color in colors_map.items():
if key in ind.upper():
bar_colors.append(color)
break
else:
bar_colors.append('#95a5a6')
# Bar chart
x = range(len(indicators))
bars = ax1.bar(x, zile, color=bar_colors, alpha=0.8)
ax1.set_xticks(x)
ax1.set_xticklabels([ind[:20] for ind in indicators], rotation=45, ha='right', fontsize=9)
ax1.set_ylabel('Zile')
ax1.set_title('Ciclu Conversie Cash - Componente')
# Add value labels on bars
for bar, val in zip(bars, zile):
height = bar.get_height()
ax1.text(bar.get_x() + bar.get_width()/2., height,
f'{int(val)}',
ha='center', va='bottom', fontsize=10, fontweight='bold')
# Waterfall-style visualization
# DIO + DSO - DPO = CCC
dio = next((z for i, z in zip(indicators, zile) if 'DIO' in i.upper()), 0)
dso = next((z for i, z in zip(indicators, zile) if 'DSO' in i.upper() and 'DIO' not in i.upper()), 0)
dpo = next((z for i, z in zip(indicators, zile) if 'DPO' in i.upper()), 0)
ccc = dio + dso - dpo
waterfall_labels = ['DIO\n(Zile Stoc)', 'DSO\n(Zile Incasare)', 'DPO\n(Zile Plata)', 'CCC\n(Ciclu Total)']
waterfall_values = [dio, dso, -dpo, ccc]
waterfall_colors = ['#3498db', '#e74c3c', '#2ecc71', '#9b59b6']
# Calculate positions for waterfall
cumulative = [0]
for i, v in enumerate(waterfall_values[:-1]):
cumulative.append(cumulative[-1] + v)
ax2.bar([0, 1, 2], [dio, dso, dpo], color=['#3498db', '#e74c3c', '#2ecc71'], alpha=0.8)
ax2.axhline(y=ccc, color='#9b59b6', linewidth=3, linestyle='--', label=f'CCC = {int(ccc)} zile')
ax2.set_xticks([0, 1, 2])
ax2.set_xticklabels(['DIO\n(+Stoc)', 'DSO\n(+Incasare)', 'DPO\n(-Plata)'], fontsize=9)
ax2.set_ylabel('Zile')
ax2.set_title('Formula: DIO + DSO - DPO = CCC')
ax2.legend(loc='upper right')
# Add annotation explaining the result
if ccc > 60:
verdict = "Ciclu lung - capital blocat mult timp"
verdict_color = '#c0392b'
elif ccc > 30:
verdict = "Ciclu moderat - poate fi optimizat"
verdict_color = '#d68910'
else:
verdict = "Ciclu eficient - capital rotit rapid"
verdict_color = '#27ae60'
ax2.text(0.5, -0.15, verdict, transform=ax2.transAxes,
ha='center', fontsize=10, color=verdict_color, fontweight='bold')
plt.tight_layout()
return fig

20
requirements.txt Normal file
View File

@@ -0,0 +1,20 @@
# Data Intelligence Report - Dependencies
# Install with: pip install -r requirements.txt --break-system-packages
# Oracle Database connection
oracledb>=1.4.0
# Data processing
pandas>=2.0.0
# Excel generation
openpyxl>=3.1.0
# PDF generation
reportlab>=4.0.0
# Charts and visualization
matplotlib>=3.7.0
# Environment configuration
python-dotenv>=1.0.0

6
run.bat Normal file
View File

@@ -0,0 +1,6 @@
@echo off
REM Run Data Intelligence Report Generator (Windows)
cd /d "%~dp0"
call .venv\Scripts\activate.bat
python main.py %*

9
run.sh Normal file
View File

@@ -0,0 +1,9 @@
#!/bin/bash
# Run Data Intelligence Report Generator
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$SCRIPT_DIR"
# Activate virtual environment and run
source .venv/bin/activate
python main.py "$@"