import os import re import time import openpyxl from pathlib import Path from typing import List, Dict, Any, Optional import threading class TrackerManager: # Known social media patterns and metadata SOSMED_PLATFORMS = { "instagram.com": { "name": "Instagram", "tier": "Tier 1", "badge": "Popular", "icon": "fab fa-instagram", "color": "#E1306C", "post_type": "Photo / Flyer & Caption" }, "facebook.com": { "name": "Facebook", "tier": "Tier 1", "badge": "Popular", "icon": "fab fa-facebook", "color": "#1877F2", "post_type": "Post / Flyer & Story" }, "pinterest.com": { "name": "Pinterest", "tier": "Tier 1", "badge": "Popular", "icon": "fab fa-pinterest", "color": "#E60023", "post_type": "Pin & Board" }, "twitter.com": { "name": "Twitter / X", "tier": "Tier 1", "badge": "Popular", "icon": "fab fa-x-twitter", "color": "#1DA1F2", "post_type": "Tweet / Media & Text" }, "x.com": { "name": "Twitter / X", "tier": "Tier 1", "badge": "Popular", "icon": "fab fa-x-twitter", "color": "#1DA1F2", "post_type": "Tweet / Media & Text" }, "threads.net": { "name": "Threads", "tier": "Tier 1", "badge": "Popular", "icon": "fab fa-threads", "color": "#FFFFFF", "post_type": "Thread Post" }, "tiktok.com": { "name": "TikTok", "tier": "Tier 1", "badge": "Popular", "icon": "fab fa-tiktok", "color": "#FE2C55", "post_type": "Photo Mode / Video" }, "reddit.com": { "name": "Reddit", "tier": "Tier 1", "badge": "Popular", "icon": "fab fa-reddit", "color": "#FF4500", "post_type": "Community Post" }, "tumblr.com": { "name": "Tumblr", "tier": "Tier 1", "badge": "Popular", "icon": "fab fa-tumblr", "color": "#36465D", "post_type": "Photo Post" }, "flickr.com": { "name": "Flickr", "tier": "Tier 2", "badge": "Photo", "icon": "fab fa-flickr", "color": "#FF0084", "post_type": "Photo Gallery" }, "behance.net": { "name": "Behance", "tier": "Tier 2", "badge": "Portfolio", "icon": "fab fa-behance", "color": "#1769FF", "post_type": "Creative Project" }, "dribbble.com": { "name": "Dribbble", "tier": "Tier 2", "badge": "Portfolio", "icon": "fab fa-dribbble", "color": "#EA4C89", "post_type": "Design Shot" }, "deviantart.com": { "name": "DeviantArt", "tier": "Tier 2", "badge": "Art", "icon": "fab fa-deviantart", "color": "#05CC47", "post_type": "Visual Art Post" }, "pixelfed.social": { "name": "Pixelfed", "tier": "Tier 2", "badge": "Fediverse", "icon": "fas fa-camera-retro", "color": "#6366F1", "post_type": "Decentralized Photo" }, "bsky.app": { "name": "Bluesky", "tier": "Tier 2", "badge": "Fediverse", "icon": "fas fa-cloud", "color": "#0285FF", "post_type": "Microblog Post" }, "medium.com": { "name": "Medium", "tier": "Tier 3", "badge": "Article", "icon": "fab fa-medium", "color": "#00AB6C", "post_type": "Article / Story" }, "kaskus.co.id": { "name": "Kaskus", "tier": "Tier 3", "badge": "Community", "icon": "fas fa-comments", "color": "#0099FF", "post_type": "Thread Community" }, "plurk.com": { "name": "Plurk", "tier": "Tier 3", "badge": "Microblog", "icon": "fas fa-stream", "color": "#CF682F", "post_type": "Timeline Post" } } def __init__(self, excel_path: Optional[str] = None): if excel_path: self.file_path = Path(excel_path) else: self.file_path = Path(__file__).resolve().parent / "tracker-posting.xlsx" self._ensure_tracker_file() self._lock = threading.Lock() self._cached_sites: Optional[List[Dict[str, Any]]] = None self._cached_campaigns: Optional[List[str]] = None self._cache_timestamp: float = 0 self._cache_ttl_seconds: float = 60.0 def _ensure_tracker_file(self): """Ensure tracker-posting.xlsx exists; if missing, initialize from clean template.""" if not self.file_path.exists(): template_path = self.file_path.parent / "tracker-posting.template.xlsx" if template_path.exists(): import shutil try: shutil.copyfile(template_path, self.file_path) except Exception as e: print(f"[WARN] Failed to copy tracker template: {e}") def _normalize_domain(self, raw_url: str) -> str: url = str(raw_url).strip().lower() url = re.sub(r"^https?://", "", url) url = re.sub(r"^www\.", "", url) domain = url.split("/")[0] return domain def invalidate_cache(self): """Invalidate in-memory cache when workbook is updated.""" with self._lock: self._cached_sites = None self._cached_campaigns = None self._cache_timestamp = 0 def get_campaign_columns(self, force_refresh: bool = False) -> List[str]: """Get all column headers for campaigns (Column B and onwards) with caching.""" with self._lock: now = time.time() if not force_refresh and self._cached_campaigns is not None and (now - self._cache_timestamp) < self._cache_ttl_seconds: return list(self._cached_campaigns) wb = openpyxl.load_workbook(self.file_path, data_only=True) sheet = wb["Tracker"] campaigns = [] for col in range(2, sheet.max_column + 1): val = sheet.cell(row=1, column=col).value if val: campaigns.append(str(val).strip()) self._cached_campaigns = campaigns self._cache_timestamp = now return list(campaigns) def get_sosmed_sites(self, force_refresh: bool = False) -> List[Dict[str, Any]]: """ Scan tracker-posting.xlsx and return all filtered social media sites with their row index and current campaign links. Cached in memory for high performance. """ with self._lock: now = time.time() if not force_refresh and self._cached_sites is not None and (now - self._cache_timestamp) < self._cache_ttl_seconds: return [dict(s) for s in self._cached_sites] wb = openpyxl.load_workbook(self.file_path, data_only=True) sheet = wb["Tracker"] headers = [] for col in range(1, sheet.max_column + 1): val = sheet.cell(row=1, column=col).value headers.append(str(val).strip() if val else f"Col_{col}") sosmed_sites = [] for row in range(2, sheet.max_row + 1): raw_site = sheet.cell(row=row, column=1).value if not raw_site: continue domain = self._normalize_domain(raw_site) matched_meta = None # Exact domain match or proper subdomain match for target_domain, meta in self.SOSMED_PLATFORMS.items(): if domain == target_domain or domain.endswith("." + target_domain): matched_meta = (target_domain, meta) break if matched_meta: target_domain, meta = matched_meta # Read current links in each campaign column campaign_links = {} for col_idx, col_name in enumerate(headers[1:], start=2): cell_val = sheet.cell(row=row, column=col_idx).value campaign_links[col_name] = str(cell_val).strip() if cell_val else None sosmed_sites.append({ "row_idx": row, "raw_site": str(raw_site).strip(), "domain": target_domain, "name": meta["name"], "tier": meta["tier"], "badge": meta["badge"], "icon": meta["icon"], "color": meta["color"], "post_type": meta["post_type"], "campaigns": campaign_links, "latest_link": next((v for v in reversed(list(campaign_links.values())) if v), None) }) # Sort by Tier 1 first, then by name sosmed_sites.sort(key=lambda s: (s["tier"], s["name"])) self._cached_sites = sosmed_sites self._cache_timestamp = now return [dict(s) for s in sosmed_sites] def update_campaign_link(self, row_idx: int, campaign_name: str, post_url: str) -> bool: """ Update the specified campaign column for row_idx with the post_url. Saves changes to tracker-posting.xlsx. """ with self._lock: wb = openpyxl.load_workbook(self.file_path) sheet = wb["Tracker"] # Locate or create the campaign column target_col = None for col in range(2, sheet.max_column + 1): header = sheet.cell(row=1, column=col).value if header and str(header).strip().lower() == campaign_name.strip().lower(): target_col = col break # If column doesn't exist, append new column if target_col is None: target_col = sheet.max_column + 1 sheet.cell(row=1, column=target_col).value = campaign_name.strip() # Set value sheet.cell(row=row_idx, column=target_col).value = post_url.strip() # Save workbook wb.save(self.file_path) self._cached_sites = None self._cached_campaigns = None return True def add_new_campaign_column(self, campaign_name: str) -> int: """Add a new campaign header column to the tracker.""" with self._lock: wb = openpyxl.load_workbook(self.file_path) sheet = wb["Tracker"] # Check if already exists for col in range(2, sheet.max_column + 1): header = sheet.cell(row=1, column=col).value if header and str(header).strip().lower() == campaign_name.strip().lower(): return col new_col = sheet.max_column + 1 sheet.cell(row=1, column=new_col).value = campaign_name.strip() wb.save(self.file_path) self._cached_sites = None self._cached_campaigns = None return new_col def add_website( self, domain: str, name: Optional[str] = None, tier: str = "Tier 2", post_type: str = "Social / Media Post", icon: str = "fas fa-globe", color: str = "#6366F1" ) -> Dict[str, Any]: """ Dynamically register a new website to the tracker and append it to tracker-posting.xlsx. """ norm_domain = self._normalize_domain(domain) platform_name = (name or norm_domain.split(".")[0]).capitalize() # Register in SOSMED_PLATFORMS dictionary self.SOSMED_PLATFORMS[norm_domain] = { "name": platform_name, "tier": tier, "badge": "Custom", "icon": icon, "color": color, "post_type": post_type } with self._lock: wb = openpyxl.load_workbook(self.file_path) sheet = wb["Tracker"] # Check if domain already exists in column 1 existing_row = None for r in range(2, sheet.max_row + 1): val = sheet.cell(row=r, column=1).value if val and self._normalize_domain(val) == norm_domain: existing_row = r break if existing_row is None: new_row = sheet.max_row + 1 sheet.cell(row=new_row, column=1).value = f"https://www.{norm_domain}" row_idx = new_row else: row_idx = existing_row wb.save(self.file_path) self._cached_sites = None self._cached_campaigns = None return { "domain": norm_domain, "name": platform_name, "tier": tier, "row_idx": row_idx, "post_type": post_type } def export_to_csv(self) -> str: """ Export current tracker sheet to CSV formatted string. """ import io import csv with self._lock: wb = openpyxl.load_workbook(self.file_path, data_only=True) sheet = wb["Tracker"] output = io.StringIO() writer = csv.writer(output) for row in sheet.iter_rows(values_only=True): if any(row): writer.writerow([v if v is not None else "" for v in row]) return output.getvalue() def export_to_excel_bytes(self) -> bytes: """ Export tracker workbook as raw bytes. """ with open(self.file_path, "rb") as f: return f.read()