sst_filter.py

#!/usr/bin/env python3
"""
SST persona filtering core module.
Contains the LinkedInFilter class and all filtering logic, so main.py can be a thin CLI entrypoint.
"""

import csv
import re
import json
import argparse  # kept for potential future reuse in module-level utilities
import yaml
import collections
import unicodedata
import io
from pathlib import Path
from typing import Dict, List, Tuple, Optional, Any
import logging

# Configure logging for this module
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class LinkedInFilter:
    """Main class for filtering LinkedIn connections based on persona configuration."""

    def __init__(self, config_path: str = "config.yaml"):
        self.config = self._load_config(config_path)
        self.active_persona = self.config['active_persona']
        self.persona_config = self.config['personas'][self.active_persona]
        # Default mode comes from config; CLI may override in main.py
        self.mode = self.config.get('matching', {}).get('default_mode', 'sst_only')

        if self.config['debug']['enabled']:
            level = getattr(logging, self.config['debug']['log_level'])
            logger.setLevel(level)

        self.enterprise_list = self.build_enterprise_list()
        logger.info(f"Initialized with persona: {self.active_persona}")
        logger.info(f"Persona: {self.persona_config['name']}")

    def _load_config(self, config_path: str) -> Dict[str, Any]:
        try:
            with open(config_path, 'r', encoding='utf-8') as f:
                config = yaml.safe_load(f)
            logger.info(f"Configuration loaded from {config_path}")
            return config
        except FileNotFoundError:
            logger.error(f"Configuration file not found: {config_path}")
            raise
        except yaml.YAMLError as e:
            logger.error(f"Error parsing YAML configuration: {e}")
            raise

    def normalize_text(self, text: str) -> str:
        if not text:
            return ""
        if self.config['language_support']['remove_accents']:
            text = unicodedata.normalize('NFD', text)
            text = ''.join(char for char in text if unicodedata.category(char) != 'Mn')
        if self.config['language_support']['lowercase']:
            text = text.lower()
        if self.config['language_support']['normalize_whitespace']:
            text = re.sub(r'\s+', ' ', text.strip())
        return text

    def extract_company_tokens(self, company_name: str) -> set:
        normalized = self.normalize_text(company_name)
        suffixes = r'\b(inc|llc|ltd|limited|plc|ag|sa|nv|se|co|corp|corporation|company|ltda|s\.?a\.?)\b'
        normalized = re.sub(suffixes, '', normalized)
        normalized = re.sub(r'[^\w\s]', ' ', normalized)
        tokens = [t for t in normalized.split() if len(t) > 2 and t not in {'and', 'the', 'group', 'global'}]
        return set(tokens)

    def is_innovation_role(self, position: str, company_tokens: set) -> Tuple[bool, Optional[str], List[str], List[str]]:
        """General persona matching: (is_match, primary_category, matched_keywords, matched_categories)."""
        if not position:
            return False, None, [], []

        pos_normalized = self.normalize_text(position)
        filtered_position = ' '.join(word for word in pos_normalized.split() if word not in company_tokens)

        # Exclusions
        exclude_patterns = self.config['validation']['exclude_patterns']
        for pattern in exclude_patterns:
            if pattern.lower() in filtered_position:
                logger.debug(f"Position excluded due to pattern '{pattern}': {position}")
                return False, None, [], []

        # Category matches
        keywords_config = self.persona_config['keywords']
        category_hits: Dict[str, List[str]] = {}
        total_hits = 0
        for category, patterns in keywords_config.items():
            matches_for_cat: List[str] = []
            for pattern in patterns:
                pat = pattern
                if not pat.startswith('^') and '.*' not in pat:
                    pat = r'\b' + re.escape(pat) + r'\b'
                if re.search(pat, filtered_position, re.IGNORECASE):
                    matches_for_cat.append(pattern)
            if matches_for_cat:
                category_hits[category] = matches_for_cat
                total_hits += len(matches_for_cat)

        if not category_hits:
            return False, None, [], []

        # Multi-signal rules
        cfg = self.config.get('matching', {})
        min_total = int(cfg.get('min_total_matches', cfg.get('min_keyword_match', 1)))
        required_groups = set(cfg.get('required_groups', []))
        low_signal_groups = set(cfg.get('low_signal_groups', []))
        allow_low_signal_only = bool(cfg.get('allow_low_signal_only', False))

        cats = set(category_hits.keys())
        high_signal_hit = True if not required_groups else bool(cats & required_groups)
        low_only = bool(cats) and cats.issubset(low_signal_groups) if low_signal_groups else False

        if total_hits < min_total:
            return False, None, [], []
        if not high_signal_hit:
            return False, None, [], []
        if low_only and not allow_low_signal_only:
            return False, None, [], []

        # Primary category by weight
        weights = self.config['matching'].get('category_weights', {})
        def cat_score(cat: str) -> Tuple[float, int]:
            return (float(weights.get(cat, 0.5)), len(category_hits.get(cat, [])))
        primary_category = max(category_hits.keys(), key=lambda c: cat_score(c))

        matched_categories = list(category_hits.keys())
        matched_keywords: List[str] = []
        for _, kws in category_hits.items():
            matched_keywords.extend(kws)

        return True, primary_category, matched_keywords, matched_categories

    def build_enterprise_list(self) -> List[str]:
        enterprises = []
        enterprise_config = self.config['enterprise_companies']
        for _, companies in enterprise_config.items():
            enterprises.extend([self.normalize_text(company) for company in companies])
        logger.info(f"Built enterprise list with {len(enterprises)} companies")
        return enterprises

    def is_enterprise_company(self, company: str) -> Tuple[bool, Optional[str]]:
        if not company:
            return False, None
        company_normalized = self.normalize_text(company)
        for enterprise in self.enterprise_list:
            pattern = r'\b' + re.escape(enterprise) + r'\b'
            if re.search(pattern, company_normalized):
                logger.debug(f"Enterprise match found: {company} -> {enterprise}")
                return True, enterprise
        return False, None

    def determine_seniority(self, position: str) -> str:
        position_normalized = self.normalize_text(position)
        seniority_config = self.config['matching']['seniority_levels']
        for level, keywords in seniority_config.items():
            for keyword in keywords:
                if keyword.lower() in position_normalized:
                    return level
        return 'specialist'

    def calculate_match_confidence(self, primary_category: str, matched_keywords: List[str]) -> float:
        if not primary_category:
            return 0.0
        category_weights = self.config['matching']['category_weights']
        base_confidence = category_weights.get(primary_category, 0.5)
        keyword_boost = min(len(matched_keywords) * 0.1, 0.3)
        return round(min(base_confidence + keyword_boost, 1.0), 3)

    def read_linkedin_csv(self, input_path: str) -> csv.DictReader:
        try:
            with open(input_path, 'r', encoding='utf-8-sig') as f:
                lines = f.readlines()
        except UnicodeDecodeError:
            with open(input_path, 'r', encoding='utf-8') as f:
                lines = f.readlines()
        header_line_idx = None
        for i, line in enumerate(lines):
            if line.startswith('First Name,'):
                header_line_idx = i
                break
        if header_line_idx is None:
            raise ValueError("Could not find CSV headers starting with 'First Name,'")
        csv_data = ''.join(lines[header_line_idx:])
        return csv.DictReader(io.StringIO(csv_data))

    def process_connections(self, input_path: str) -> List[Dict[str, Any]]:
        logger.info(f"Processing connections from {input_path}")
        reader = self.read_linkedin_csv(input_path)
        results: List[Dict[str, Any]] = []
        stats = {
            'total_connections': 0,
            'persona_roles': 0,
            'enterprise_companies': 0,
            'final_matches': 0,
            'categories': collections.defaultdict(int),
            'seniority_levels': collections.defaultdict(int),
            'companies': collections.defaultdict(int)
        }
        for row in reader:
            stats['total_connections'] += 1
            first_name = row.get('First Name', '').strip()
            last_name = row.get('Last Name', '').strip()
            company = row.get('Company', '').strip()
            position = row.get('Position', '').strip()
            if not all([first_name, last_name, company, position]):
                logger.debug("Skipping row with missing required fields")
                continue
            company_tokens = self.extract_company_tokens(company)
            is_match, primary_category, matched_keywords, matched_categories = self.is_innovation_role(position, company_tokens)
            if is_match:
                stats['persona_roles'] += 1
                stats['categories'][primary_category] += 1
            is_enterprise, enterprise_match = self.is_enterprise_company(company)
            if is_enterprise:
                stats['enterprise_companies'] += 1
            passes_gate = is_match if self.mode == 'sst_only' else (is_match and is_enterprise)
            if passes_gate:
                seniority = self.determine_seniority(position)
                confidence = self.calculate_match_confidence(primary_category, matched_keywords)
                result = {
                    'First Name': first_name,
                    'Last Name': last_name,
                    'Company': company,
                    'Position': position,
                    'URL': row.get('URL', ''),
                    'Email Address': row.get('Email Address', ''),
                    'Connected On': row.get('Connected On', ''),
                    'match_primary_category': primary_category,
                    'matched_categories': '|'.join(matched_categories),
                    'matched_keywords': '|'.join(matched_keywords),
                    'enterprise_match': enterprise_match or '',
                    'match_confidence': confidence,
                    'seniority_level': seniority,
                    'persona': self.active_persona
                }
                results.append(result)
                stats['final_matches'] += 1
                stats['seniority_levels'][seniority] += 1
                stats['companies'][company] += 1
        logger.info(f"Processing complete: {stats['final_matches']} matches from {stats['total_connections']} connections")
        return results, dict(stats)

    def save_results(self, results: List[Dict[str, Any]], stats: Dict[str, Any], output_dir: Path) -> None:
        output_dir.mkdir(exist_ok=True)
        persona_name = self.active_persona
        results_filename = f"{persona_name}_{self.config['output']['results_filename']}"
        summary_filename = f"{persona_name}_{self.config['output']['summary_filename']}"
        if results:
            results_path = output_dir / results_filename
            with open(results_path, 'w', newline='', encoding='utf-8') as f:
                writer = csv.DictWriter(f, fieldnames=results[0].keys())
                writer.writeheader()
                writer.writerows(results)
            logger.info(f"Results saved to {results_path}")
        summary = {
            'persona': {
                'name': self.persona_config['name'],
                'description': self.persona_config['description'],
                'active_persona': self.active_persona
            },
            'statistics': stats,
            'match_rates': {
                'persona_rate': f"{(stats['final_matches']/stats['total_connections']*100):.1f}%" if stats['total_connections'] > 0 else "0%",
                'enterprise_rate': f"{(stats['enterprise_companies']/stats['total_connections']*100):.1f}%" if stats['total_connections'] > 0 else "0%"
            },
            'top_companies': dict(collections.Counter(stats['companies']).most_common(10)),
            'configuration_used': {
                'fuzzy_threshold': self.config['matching']['fuzzy_threshold'],
                'min_keyword_match': self.config['matching']['min_keyword_match'],
                'category_weights': self.config['matching']['category_weights'],
                'mode': self.mode
            }
        }
        summary_path = output_dir / summary_filename
        with open(summary_path, 'w', encoding='utf-8') as f:
            json.dump(summary, f, indent=2, ensure_ascii=False)
        logger.info(f"Summary saved to {summary_path}")

    def display_results(self, results: List[Dict[str, Any]], stats: Dict[str, Any]) -> None:
        print(f"\n{'='*80}")
        print(f"LINKEDIN CONNECTION FILTER RESULTS")
        print(f"Persona: {self.persona_config['name']}")
        print(f"{'='*80}")
        print(f"\nSUMMARY STATISTICS:")
        print(f"Total connections processed: {stats['total_connections']}")
        print(f"SST-related matches: {stats['final_matches']} ({(stats['final_matches']/stats['total_connections']*100):.1f}%)")
        print(f"Enterprise company connections: {stats['enterprise_companies']} ({(stats['enterprise_companies']/stats['total_connections']*100):.1f}%)")
        if stats['categories']:
            print(f"\nCATEGORIES:")
            for category, count in stats['categories'].items():
                print(f"  {category}: {count}")
        if stats['seniority_levels']:
            print(f"\nSENIORITY LEVELS:")
            for level, count in stats['seniority_levels'].items():
                print(f"  {level}: {count}")
        if results:
            print(f"\nTOP MATCHED CONNECTIONS:")
            print("-" * 120)
            for i, result in enumerate(results[:10]):
                print(f"{i+1:2d}. {result['First Name']} {result['Last Name']}")
                print(f"    Position: {result['Position']}")
                print(f"    Company: {result['Company']}")
                print(f"    Category: {result['match_primary_category']} | Confidence: {result['match_confidence']} | Seniority: {result['seniority_level']}")
                print(f"    Categories: {result['matched_categories']}")
                print(f"    Keywords: {result['matched_keywords']}")
                if result['Email Address']:
                    print(f"    Email: {result['Email Address']}")
                print(f"    LinkedIn: {result['URL']}")
                print("-" * 120)
            if len(results) > 10:
                print(f"... and {len(results) - 10} more matches (see CSV file for complete results)")