Floke/Brancheneinstufung2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(classification): restore service logic and standardization formula

Browse Source 
- Restored missing method implementations in ClassificationService (classify, extract_metrics)
- Fixed Standardization Logic not being applied in metric cascade
- Bumped version to v0.7.4 in config.py
- Removed duplicate API endpoint in app.py
- Updated MIGRATION_PLAN.md
This commit is contained in:
Floke
2026-01-24 13:34:04 +00:00
parent 01985aca33
commit 31e1a5fc08
4 changed files with 140 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
company-explorer/backend/app.py
View File

@@ -342,47 +342,6 @@ def override_impressum(company_id: int, url: str, background_tasks: BackgroundTa
db.commit()
return {"status": "updated"}
@app.get("/api/companies/export")
def export_companies_csv(db: Session = Depends(get_db)):
"""
Exports a CSV of all companies with their key metrics.
"""
import io
import csv
from fastapi.responses import StreamingResponse
output = io.StringIO()
writer = csv.writer(output)
# Header
writer.writerow([
"ID", "Name", "Website", "City", "Country", "AI Industry",
"Metric Name", "Metric Value", "Metric Unit", "Standardized Value (m2)",
"Source", "Source URL", "Confidence", "Proof Text"
])
companies = db.query(Company).order_by(Company.name.asc()).all()
for c in companies:
writer.writerow([
c.id, c.name, c.website, c.city, c.country, c.industry_ai,
c.calculated_metric_name,
c.calculated_metric_value,
c.calculated_metric_unit,
c.standardized_metric_value,
c.metric_source,
c.metric_source_url,
c.metric_confidence,
c.metric_proof_text
])
output.seek(0)
return StreamingResponse(
output,
media_type="text/csv",
headers={"Content-Disposition": f"attachment; filename=company_export_{datetime.utcnow().strftime('%Y-%m-%d')}.csv"}
)
def run_wikipedia_reevaluation_task(company_id: int):

4
company-explorer/backend/config.py
View File

@@ -10,7 +10,7 @@ try:
class Settings(BaseSettings):
# App Info
APP_NAME: str = "Company Explorer"
VERSION: str = "0.6.4"
VERSION: str = "0.7.3"
DEBUG: bool = True
# Database (FINAL CORRECT PATH for Docker Container)
@@ -34,7 +34,7 @@ except ImportError:
# Fallback wenn pydantic-settings nicht installiert ist
class FallbackSettings:
APP_NAME = "Company Explorer"
VERSION = "0.6.4"
VERSION = "0.7.3"
DEBUG = True
DATABASE_URL = "sqlite:////app/companies_v3_fixed_2.db" # FINAL CORRECT PATH
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")

143
company-explorer/backend/services/classification.py
View File

@@ -32,16 +32,80 @@ class ClassificationService:
return enrichment.content if enrichment and enrichment.content else None
def _run_llm_classification_prompt(self, website_text: str, company_name: str, industry_definitions: List[Dict[str, str]]) -> Optional[str]:
# ... [omitted for brevity, no changes here] ...
pass
prompt = f"""
Act as a strict B2B Industry Classifier.
Company: {company_name}
Context: {website_text[:3000]}
Available Industries:
{json.dumps(industry_definitions, indent=2)}
Task: Select the ONE industry that best matches the company.
If the company is a Hospital/Klinik, select 'Healthcare - Hospital'.
If none match well, select 'Others'.
Return ONLY the exact name of the industry.
"""
try:
response = call_gemini_flash(prompt)
if not response: return "Others"
cleaned = response.strip().replace('"', '').replace("'", "")
# Simple fuzzy match check
valid_names = [i['name'] for i in industry_definitions] + ["Others"]
if cleaned in valid_names:
return cleaned
# Fallback: Try to find name in response
for name in valid_names:
if name in cleaned:
return name
return "Others"
except Exception as e:
logger.error(f"Classification Prompt Error: {e}")
return "Others"
def _run_llm_metric_extraction_prompt(self, text_content: str, search_term: str, industry_name: str) -> Optional[Dict[str, Any]]:
# ... [omitted for brevity, no changes here] ...
pass
prompt = f"""
Extract the following metric for the company in industry '{industry_name}':
Target Metric: "{search_term}"
Source Text:
{text_content[:6000]}
Return a JSON object with:
- "raw_value": The number found (e.g. 352 or 352.0). If text says "352 Betten", extract 352. If not found, null.
- "raw_unit": The unit found (e.g. "Betten", "m²").
- "proof_text": A short quote from the text proving this value.
JSON ONLY.
"""
try:
response = call_gemini_flash(prompt, json_mode=True)
if not response: return None
if isinstance(response, str):
response = response.replace("```json", "").replace("```", "").strip()
data = json.loads(response)
else:
data = response
# Basic cleanup
if data.get("raw_value") == "null": data["raw_value"] = None
return data
except Exception as e:
logger.error(f"LLM Extraction Parse Error: {e}")
return None
def _is_metric_plausible(self, metric_name: str, value: Optional[float]) -> bool:
# ... [omitted for brevity, no changes here] ...
pass
if value is None: return False
try:
val_float = float(value)
return val_float > 0
except:
return False
def _parse_standardization_logic(self, formula: str, raw_value: float) -> Optional[float]:
if not formula or raw_value is None:
@@ -104,12 +168,35 @@ class ClassificationService:
all_source_results.append((source_name, llm_result))
except Exception as e:
logger.error(f"Error in {source_name} stage: {e}")
processed_results = []
# ... [processing logic as before, no changes] ...
for source_name, llm_result in all_source_results:
metric_value = llm_result.get("raw_value")
metric_unit = llm_result.get("raw_unit")
if metric_value is not None and self._is_metric_plausible(search_term, metric_value):
standardized_value = None
if standardization_logic and metric_value is not None:
standardized_value = self._parse_standardization_logic(standardization_logic, metric_value)
processed_results.append({
"calculated_metric_name": search_term,
"calculated_metric_value": metric_value,
"calculated_metric_unit": metric_unit,
"standardized_metric_value": standardized_value,
"standardized_metric_unit": standardized_unit,
"metric_source": source_name,
"metric_proof_text": llm_result.get("proof_text"),
"metric_source_url": llm_result.get("source_url"),
"metric_confidence": 0.95,
"metric_confidence_reason": "Value found and extracted by LLM."
})
else:
logger.info(f"LLM found no plausible metric for {search_term} in {source_name}.")
best_result = self._get_best_metric_result(processed_results)
return best_result if best_result else final_result
# ... [rest of the class, no changes] ...
def extract_metrics_for_industry(self, company: Company, db: Session, industry: Industry) -> Company:
if not industry or not industry.scraper_search_term:
logger.warning(f"No metric configuration for industry '{industry.name if industry else 'None'}'")
@@ -141,9 +228,41 @@ class ClassificationService:
return company
def reevaluate_wikipedia_metric(self, company: Company, db: Session, industry: Industry) -> Company:
# ... [omitted for brevity, no changes here] ...
pass
logger.info(f"Re-evaluating metric for {company.name}...")
return self.extract_metrics_for_industry(company, db, industry)
def classify_company_potential(self, company: Company, db: Session) -> Company:
# ... [omitted for brevity, no changes here] ...
pass
logger.info(f"Starting classification for {company.name}...")
# 1. Load Definitions
industries = self._load_industry_definitions(db)
industry_defs = [{"name": i.name, "description": i.description} for i in industries]
# 2. Get Content (Website)
website_content, _ = self._get_website_content_and_url(company)
if not website_content:
logger.warning(f"No website content for {company.name}. Skipping classification.")
return company
# 3. Classify Industry
suggested_industry_name = self._run_llm_classification_prompt(website_content, company.name, industry_defs)
logger.info(f"AI suggests industry: {suggested_industry_name}")
# 4. Update Company
# Match back to DB object
matched_industry = next((i for i in industries if i.name == suggested_industry_name), None)
if matched_industry:
company.industry_ai = matched_industry.name
else:
company.industry_ai = "Others"
# 5. Extract Metrics (Cascade)
if matched_industry:
self.extract_metrics_for_industry(company, db, matched_industry)
company.last_classification_at = datetime.utcnow()
db.commit()
return company