Refactor SEO automation into unified CLI application
Major refactoring to create a clean, integrated CLI application: ### New Features: - Unified CLI executable (./seo) with simple command structure - All commands accept optional CSV file arguments - Auto-detection of latest files when no arguments provided - Simplified output directory structure (output/ instead of output/reports/) - Cleaner export filename format (all_posts_YYYY-MM-DD.csv) ### Commands: - export: Export all posts from WordPress sites - analyze [csv]: Analyze posts with AI (optional CSV input) - recategorize [csv]: Recategorize posts with AI - seo_check: Check SEO quality - categories: Manage categories across sites - approve [files]: Review and approve recommendations - full_pipeline: Run complete workflow - analytics, gaps, opportunities, report, status ### Changes: - Moved all scripts to scripts/ directory - Created config.yaml for configuration - Updated all scripts to use output/ directory - Deprecated old seo-cli.py in favor of new ./seo - Added AGENTS.md and CHANGELOG.md documentation - Consolidated README.md with updated usage ### Technical: - Added PyYAML dependency - Removed hardcoded configuration values - All scripts now properly integrated - Better error handling and user feedback Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
Kevin Bataille
352
scripts/user_approval.py
Normal file
352
scripts/user_approval.py
Normal file
@@ -0,0 +1,352 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
User Approval Mechanism for SEO Recommendations
|
||||
Allows users to review and approve recommendations from CSV files.
|
||||
"""
|
||||
|
||||
import csv
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
from datetime import datetime
|
||||
from config import Config
|
||||
|
||||
# Setup logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class UserApprovalSystem:
|
||||
"""System for reviewing and approving SEO recommendations."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the approval system."""
|
||||
self.output_dir = Path(__file__).parent.parent / 'output'
|
||||
self.approved_recommendations = []
|
||||
self.rejected_recommendations = []
|
||||
self.pending_recommendations = []
|
||||
|
||||
def load_recommendations_from_csv(self, csv_file: str) -> List[Dict]:
|
||||
"""Load recommendations from CSV file."""
|
||||
recommendations = []
|
||||
|
||||
if not Path(csv_file).exists():
|
||||
logger.error(f"CSV file not found: {csv_file}")
|
||||
return recommendations
|
||||
|
||||
try:
|
||||
with open(csv_file, 'r', encoding='utf-8') as f:
|
||||
reader = csv.DictReader(f)
|
||||
for row in reader:
|
||||
recommendations.append(dict(row))
|
||||
|
||||
logger.info(f"Loaded {len(recommendations)} recommendations from {csv_file}")
|
||||
return recommendations
|
||||
except Exception as e:
|
||||
logger.error(f"Error loading CSV: {e}")
|
||||
return recommendations
|
||||
|
||||
def display_recommendation(self, recommendation: Dict, index: int, total: int):
|
||||
"""Display a single recommendation for user review."""
|
||||
print(f"\n{'='*80}")
|
||||
print(f"RECOMMENDATION {index}/{total}")
|
||||
print(f"{'='*80}")
|
||||
|
||||
# Display different fields depending on the type of recommendation
|
||||
if 'post_title' in recommendation:
|
||||
print(f"Post Title: {recommendation.get('post_title', 'N/A')}")
|
||||
print(f"Post ID: {recommendation.get('post_id', 'N/A')}")
|
||||
print(f"Site: {recommendation.get('site', 'N/A')}")
|
||||
print(f"Current Categories: {recommendation.get('current_categories', 'N/A')}")
|
||||
print(f"Proposed Category: {recommendation.get('proposed_category', 'N/A')}")
|
||||
print(f"Proposed Site: {recommendation.get('proposed_site', 'N/A')}")
|
||||
print(f"Reason: {recommendation.get('reason', 'N/A')}")
|
||||
print(f"Confidence: {recommendation.get('confidence', 'N/A')}")
|
||||
print(f"Content Preview: {recommendation.get('content_preview', 'N/A')[:100]}...")
|
||||
elif 'title' in recommendation:
|
||||
print(f"Post Title: {recommendation.get('title', 'N/A')}")
|
||||
print(f"Post ID: {recommendation.get('post_id', 'N/A')}")
|
||||
print(f"Site: {recommendation.get('site', 'N/A')}")
|
||||
print(f"Decision: {recommendation.get('decision', 'N/A')}")
|
||||
print(f"Recommended Category: {recommendation.get('recommended_category', 'N/A')}")
|
||||
print(f"Reason: {recommendation.get('reason', 'N/A')}")
|
||||
print(f"Priority: {recommendation.get('priority', 'N/A')}")
|
||||
print(f"AI Notes: {recommendation.get('ai_notes', 'N/A')}")
|
||||
else:
|
||||
# Generic display for other types of recommendations
|
||||
for key, value in recommendation.items():
|
||||
print(f"{key.replace('_', ' ').title()}: {value}")
|
||||
|
||||
def get_user_choice(self) -> str:
|
||||
"""Get user's approval choice."""
|
||||
while True:
|
||||
print(f"\nOptions:")
|
||||
print(f" 'y' or 'yes' - Approve this recommendation")
|
||||
print(f" 'n' or 'no' - Reject this recommendation")
|
||||
print(f" 's' or 'skip' - Skip this recommendation for later review")
|
||||
print(f" 'q' or 'quit' - Quit and save current progress")
|
||||
|
||||
choice = input(f"\nEnter your choice: ").strip().lower()
|
||||
|
||||
if choice in ['y', 'yes']:
|
||||
return 'approved'
|
||||
elif choice in ['n', 'no']:
|
||||
return 'rejected'
|
||||
elif choice in ['s', 'skip']:
|
||||
return 'pending'
|
||||
elif choice in ['q', 'quit']:
|
||||
return 'quit'
|
||||
else:
|
||||
print("Invalid choice. Please enter 'y', 'n', 's', or 'q'.")
|
||||
|
||||
def review_recommendations(self, recommendations: List[Dict], title: str = "Recommendations"):
|
||||
"""Review recommendations with user interaction."""
|
||||
print(f"\n{'='*80}")
|
||||
print(f"REVIEWING {title.upper()}")
|
||||
print(f"Total recommendations to review: {len(recommendations)}")
|
||||
print(f"{'='*80}")
|
||||
|
||||
for i, recommendation in enumerate(recommendations, 1):
|
||||
self.display_recommendation(recommendation, i, len(recommendations))
|
||||
|
||||
choice = self.get_user_choice()
|
||||
|
||||
if choice == 'quit':
|
||||
logger.info("User chose to quit. Saving progress...")
|
||||
break
|
||||
elif choice == 'approved':
|
||||
recommendation['status'] = 'approved'
|
||||
self.approved_recommendations.append(recommendation)
|
||||
logger.info(f"Approved recommendation {i}")
|
||||
elif choice == 'rejected':
|
||||
recommendation['status'] = 'rejected'
|
||||
self.rejected_recommendations.append(recommendation)
|
||||
logger.info(f"Rejected recommendation {i}")
|
||||
elif choice == 'pending':
|
||||
recommendation['status'] = 'pending_review'
|
||||
self.pending_recommendations.append(recommendation)
|
||||
logger.info(f"Skipped recommendation {i} for later review")
|
||||
|
||||
def export_approved_recommendations(self, filename_suffix: str = "") -> str:
|
||||
"""Export approved recommendations to CSV."""
|
||||
if not self.approved_recommendations:
|
||||
logger.info("No approved recommendations to export")
|
||||
return ""
|
||||
|
||||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||
filename = f"approved_recommendations_{timestamp}{filename_suffix}.csv"
|
||||
csv_file = self.output_dir / filename
|
||||
|
||||
# Get all unique fieldnames from recommendations
|
||||
fieldnames = set()
|
||||
for rec in self.approved_recommendations:
|
||||
fieldnames.update(rec.keys())
|
||||
fieldnames = sorted(list(fieldnames))
|
||||
|
||||
with open(csv_file, 'w', newline='', encoding='utf-8') as f:
|
||||
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
||||
writer.writeheader()
|
||||
writer.writerows(self.approved_recommendations)
|
||||
|
||||
logger.info(f"Exported {len(self.approved_recommendations)} approved recommendations to: {csv_file}")
|
||||
return str(csv_file)
|
||||
|
||||
def export_rejected_recommendations(self, filename_suffix: str = "") -> str:
|
||||
"""Export rejected recommendations to CSV."""
|
||||
if not self.rejected_recommendations:
|
||||
logger.info("No rejected recommendations to export")
|
||||
return ""
|
||||
|
||||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||
filename = f"rejected_recommendations_{timestamp}{filename_suffix}.csv"
|
||||
csv_file = self.output_dir / filename
|
||||
|
||||
# Get all unique fieldnames from recommendations
|
||||
fieldnames = set()
|
||||
for rec in self.rejected_recommendations:
|
||||
fieldnames.update(rec.keys())
|
||||
fieldnames = sorted(list(fieldnames))
|
||||
|
||||
with open(csv_file, 'w', newline='', encoding='utf-8') as f:
|
||||
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
||||
writer.writeheader()
|
||||
writer.writerows(self.rejected_recommendations)
|
||||
|
||||
logger.info(f"Exported {len(self.rejected_recommendations)} rejected recommendations to: {csv_file}")
|
||||
return str(csv_file)
|
||||
|
||||
def export_pending_recommendations(self, filename_suffix: str = "") -> str:
|
||||
"""Export pending recommendations to CSV."""
|
||||
if not self.pending_recommendations:
|
||||
logger.info("No pending recommendations to export")
|
||||
return ""
|
||||
|
||||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||
filename = f"pending_recommendations_{timestamp}{filename_suffix}.csv"
|
||||
csv_file = self.output_dir / filename
|
||||
|
||||
# Get all unique fieldnames from recommendations
|
||||
fieldnames = set()
|
||||
for rec in self.pending_recommendations:
|
||||
fieldnames.update(rec.keys())
|
||||
fieldnames = sorted(list(fieldnames))
|
||||
|
||||
with open(csv_file, 'w', newline='', encoding='utf-8') as f:
|
||||
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
||||
writer.writeheader()
|
||||
writer.writerows(self.pending_recommendations)
|
||||
|
||||
logger.info(f"Exported {len(self.pending_recommendations)} pending recommendations to: {csv_file}")
|
||||
return str(csv_file)
|
||||
|
||||
def run_interactive_approval(self, csv_files: List[str]):
|
||||
"""Run interactive approval process for multiple CSV files."""
|
||||
logger.info("="*70)
|
||||
logger.info("USER APPROVAL SYSTEM FOR SEO RECOMMENDATIONS")
|
||||
logger.info("="*70)
|
||||
|
||||
for csv_file in csv_files:
|
||||
logger.info(f"\nLoading recommendations from: {csv_file}")
|
||||
recommendations = self.load_recommendations_from_csv(csv_file)
|
||||
|
||||
if not recommendations:
|
||||
logger.warning(f"No recommendations found in {csv_file}, skipping...")
|
||||
continue
|
||||
|
||||
# Get the filename without path for the title
|
||||
filename = Path(csv_file).stem
|
||||
self.review_recommendations(recommendations, title=filename)
|
||||
|
||||
# Export results
|
||||
logger.info("\n" + "="*70)
|
||||
logger.info("EXPORTING RESULTS")
|
||||
logger.info("="*70)
|
||||
|
||||
approved_file = self.export_approved_recommendations()
|
||||
rejected_file = self.export_rejected_recommendations()
|
||||
pending_file = self.export_pending_recommendations()
|
||||
|
||||
# Summary
|
||||
logger.info(f"\n{'─'*70}")
|
||||
logger.info("APPROVAL SUMMARY:")
|
||||
logger.info(f" Approved: {len(self.approved_recommendations)}")
|
||||
logger.info(f" Rejected: {len(self.rejected_recommendations)}")
|
||||
logger.info(f" Pending: {len(self.pending_recommendations)}")
|
||||
logger.info(f"{'─'*70}")
|
||||
|
||||
if approved_file:
|
||||
logger.info(f"\nApproved recommendations saved to: {approved_file}")
|
||||
if rejected_file:
|
||||
logger.info(f"Rejected recommendations saved to: {rejected_file}")
|
||||
if pending_file:
|
||||
logger.info(f"Pending recommendations saved to: {pending_file}")
|
||||
|
||||
logger.info(f"\n✓ Approval process complete!")
|
||||
|
||||
def run_auto_approval(self, csv_files: List[str], auto_approve_threshold: float = 0.8):
|
||||
"""Auto-approve recommendations based on confidence threshold."""
|
||||
logger.info("="*70)
|
||||
logger.info("AUTO APPROVAL SYSTEM FOR SEO RECOMMENDATIONS")
|
||||
logger.info("="*70)
|
||||
logger.info(f"Auto-approval threshold: {auto_approve_threshold}")
|
||||
|
||||
all_recommendations = []
|
||||
for csv_file in csv_files:
|
||||
logger.info(f"\nLoading recommendations from: {csv_file}")
|
||||
recommendations = self.load_recommendations_from_csv(csv_file)
|
||||
all_recommendations.extend(recommendations)
|
||||
|
||||
approved_count = 0
|
||||
rejected_count = 0
|
||||
|
||||
for rec in all_recommendations:
|
||||
# Check if there's a confidence field and if it meets the threshold
|
||||
confidence_str = rec.get('confidence', 'Low').lower()
|
||||
confidence_value = 0.0
|
||||
|
||||
if confidence_str == 'high':
|
||||
confidence_value = 0.9
|
||||
elif confidence_str == 'medium':
|
||||
confidence_value = 0.6
|
||||
elif confidence_str == 'low':
|
||||
confidence_value = 0.3
|
||||
else:
|
||||
# Try to parse as numeric value if possible
|
||||
try:
|
||||
confidence_value = float(confidence_str)
|
||||
except ValueError:
|
||||
confidence_value = 0.3 # Default to low
|
||||
|
||||
if confidence_value >= auto_approve_threshold:
|
||||
rec['status'] = 'auto_approved'
|
||||
self.approved_recommendations.append(rec)
|
||||
approved_count += 1
|
||||
else:
|
||||
rec['status'] = 'auto_rejected'
|
||||
self.rejected_recommendations.append(rec)
|
||||
rejected_count += 1
|
||||
|
||||
# Export results
|
||||
logger.info("\n" + "="*70)
|
||||
logger.info("EXPORTING AUTO-APPROVAL RESULTS")
|
||||
logger.info("="*70)
|
||||
|
||||
approved_file = self.export_approved_recommendations("_auto")
|
||||
rejected_file = self.export_rejected_recommendations("_auto")
|
||||
|
||||
# Summary
|
||||
logger.info(f"\n{'─'*70}")
|
||||
logger.info("AUTO APPROVAL SUMMARY:")
|
||||
logger.info(f" Auto-approved: {approved_count}")
|
||||
logger.info(f" Auto-rejected: {rejected_count}")
|
||||
logger.info(f"{'─'*70}")
|
||||
|
||||
if approved_file:
|
||||
logger.info(f"\nAuto-approved recommendations saved to: {approved_file}")
|
||||
if rejected_file:
|
||||
logger.info(f"Auto-rejected recommendations saved to: {rejected_file}")
|
||||
|
||||
logger.info(f"\n✓ Auto-approval process complete!")
|
||||
|
||||
|
||||
def main():
|
||||
"""Main entry point."""
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description='Review and approve SEO recommendations'
|
||||
)
|
||||
parser.add_argument(
|
||||
'csv_files',
|
||||
nargs='+',
|
||||
help='CSV files containing recommendations to review'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--auto',
|
||||
action='store_true',
|
||||
help='Run auto-approval mode instead of interactive mode'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--threshold',
|
||||
type=float,
|
||||
default=0.8,
|
||||
help='Confidence threshold for auto-approval (default: 0.8)'
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
approval_system = UserApprovalSystem()
|
||||
|
||||
if args.auto:
|
||||
approval_system.run_auto_approval(args.csv_files, args.threshold)
|
||||
else:
|
||||
approval_system.run_interactive_approval(args.csv_files)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user