内容生成与自动化

# AIGC 基础调用框架
import os
from openai import OpenAI
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class GenerationRequest:
    prompt: str
    model: str = "gpt-4"
    max_tokens: int = 2048
    temperature: float = 0.7
    system_prompt: Optional[str] = None

@dataclass
class GenerationResult:
    content: str
    model: str
    usage: dict
    finish_reason: str

class AIGCPipeline:
    def __init__(self, api_key: str = None):
        self.client = OpenAI(api_key=api_key or os.getenv("OPENAI_API_KEY"))
        self.history: List[dict] = []

    def generate(self, request: GenerationRequest) -> GenerationResult:
        messages = []
        if request.system_prompt:
            messages.append({"role": "system", "content": request.system_prompt})
        messages.append({"role": "user", "content": request.prompt})

        response = self.client.chat.completions.create(
            model=request.model,
            messages=messages,
            max_tokens=request.max_tokens,
            temperature=request.temperature
        )

        choice = response.choices[0]
        return GenerationResult(
            content=choice.message.content,
            model=response.model,
            usage=response.usage.model_dump(),
            finish_reason=choice.finish_reason
        )

# 提示词模板引擎
from string import Template
from typing import Dict, Any
import json

class PromptTemplate:
    def __init__(self, template: str):
        self.template = Template(template)

    def render(self, **kwargs: Any) -> str:
        return self.template.safe_substitute(**kwargs)

    def render_json(self, **kwargs: Any) -> str:
        data = self.render(**kwargs)
        return json.dumps({"prompt": data}, ensure_ascii=False)

# 预定义模板库
TEMPLATES = {
    "article_outline": PromptTemplate(
        "请为主题 "$topic" 生成文章大纲，要求包含 $n_sections 个章节，"
        "目标读者为 $audience，风格为 $style。输出 JSON 格式。"
    ),
    "content_rewrite": PromptTemplate(
        "将以下内容改写为 $tone 风格，保持核心信息不变：

$content"
    ),
    "multi_language": PromptTemplate(
        "将以下文本翻译为 $language，要求：$requirements

$text"
    ),
}

# 抽取式摘要（TextRank 算法）
import networkx as nx
from sklearn.feature_extraction.text import TfidfVectorizer
from typing import List

class TextRankSummarizer:
    def __init__(self, ratio: float = 0.2):
        self.ratio = ratio

    def _split_sentences(self, text: str) -> List[str]:
        import re
        sentences = re.split(r'(?<=[.!?。！？])s+', text.strip())
        return [s.strip() for s in sentences if len(s.strip()) > 10]

    def summarize(self, text: str) -> str:
        sentences = self._split_sentences(text)
        if len(sentences) < 3:
            return text

        vectorizer = TfidfVectorizer().fit(sentences)
        matrix = vectorizer.transform(sentences)
        similarity = (matrix * matrix.T).A

        graph = nx.Graph()
        for i in range(len(sentences)):
            graph.add_node(i)
            for j in range(i + 1, len(sentences)):
                if similarity[i, j] > 0.1:
                    graph.add_edge(i, j, weight=similarity[i, j])

        scores = nx.pagerank(graph, weight='weight')
        n = max(1, int(len(sentences) * self.ratio))
        top_indices = sorted(scores, key=scores.get, reverse=True)[:n]
        top_indices.sort()
        return ' '.join(sentences[i] for i in top_indices)

# 生成式摘要与改写（LLM）
from openai import OpenAI
import json

class LLMRewriter:
    def __init__(self):
        self.client = OpenAI()

    def summarize(self, text: str, max_words: int = 200) -> str:
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "你是一个专业的文本摘要助手。"},
                {"role": "user", "content":
                    f"请将以下文本摘要为 {max_words} 字以内的精炼摘要，"
                    f"保留核心事实和关键数据：

{text}"}
            ],
            temperature=0.3
        )
        return response.choices[0].message.content

    def rewrite(self, text: str, style: str, audience: str = "general") -> str:
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": f"将文本改写为{style}风格，"
                                              f"目标受众为{audience}。"},
                {"role": "user", "content": f"改写以下内容：

{text}"}
            ],
            temperature=0.7
        )
        return response.choices[0].message.content

    def simplify(self, text: str, grade_level: int = 8) -> str:
        return self.rewrite(text, style="通俗易懂，适合" + str(grade_level) + "年级阅读")

# 营销文案生成器（AIDA 框架）
from openai import OpenAI
from dataclasses import dataclass
from typing import List
import random

@dataclass
class BrandProfile:
    name: str
    tone: str
    target_audience: str
    banned_words: List[str]
    key_messages: List[str]

class MarketingCopyGenerator:
    def __init__(self, brand: BrandProfile):
        self.client = OpenAI()
        self.brand = brand

    def generate_aida(self, product: str, features: List[str]) -> dict:
        system_prompt = (
            f"你是{self.brand.name}的文案专家。品牌调性：{self.brand.tone}。"
            f"目标受众：{self.brand.target_audience}。"
            f"禁用词：{', '.join(self.brand.banned_words)}。"
            f"核心信息：{', '.join(self.brand.key_messages)}。"
            f"使用 AIDA 模型（Attention, Interest, Desire, Action）生成文案。"
            f"输出 JSON 格式，包含 attention、interest、desire、action 四个字段。"
        )
        user_prompt = (
            f"为产品 '{product}' 生成营销文案。
"
            f"产品特点：{', '.join(features)}"
        )
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            temperature=0.8,
            response_format={"type": "json_object"}
        )
        return response.choices[0].message.content

    def generate_variants(self, product: str, features: List[str],
                          n: int = 5) -> List[str]:
        variants = []
        for _ in range(n):
            seed = random.randint(1, 10000)
            aida = self.generate_aida(product, features)
            variants.append(aida)
        return variants

# 文案质量自动化评分
import re
from typing import Dict, List

class CopyScorer:
    def __init__(self, banned_words: List[str] = None):
        self.banned_words = banned_words or []

    def score(self, copy: str) -> Dict[str, float]:
        return {
            "readability": self._readability(copy),
            "sentiment": self._sentiment_score(copy),
            "call_to_action": self._cta_strength(copy),
            "banned_word_penalty": self._banned_penalty(copy),
            "overall": 0.0
        }

    def _readability(self, text: str) -> float:
        sentences = re.split(r'[.!?。！？]', text)
        words = text.split()
        if not sentences or not words:
            return 0.0
        avg_sent_len = len(words) / max(len(sentences), 1)
        # 理想句长 10-20 词
        score = max(0, 1.0 - abs(avg_sent_len - 15) / 15)
        return round(score * 100, 1)

    def _cta_strength(self, text: str) -> float:
        cta_keywords = [
            "立即购买", "立即体验", "马上行动", "限时优惠",
            "免费注册", "立即订阅", "Get started", "Try now"
        ]
        found = sum(1 for kw in cta_keywords if kw.lower() in text.lower())
        return min(found * 25.0, 100.0)

    def _sentiment_score(self, text: str) -> float:
        positive = ["优秀", "卓越", "创新", "领先", "高效", "卓越", "amazing"]
        negative = ["问题", "困难", "挑战", "复杂", "昂贵"]
        pos_count = sum(1 for w in positive if w in text)
        neg_count = sum(1 for w in negative if w in text)
        total = pos_count + neg_count
        if total == 0:
            return 50.0
        return round((pos_count / total) * 100, 1)

    def _banned_penalty(self, text: str) -> float:
        found = [w for w in self.banned_words if w in text]
        return max(0, 100.0 - len(found) * 20.0)

# AI 代码生成 + 静态验证流水线
import subprocess
import tempfile
import os
from openai import OpenAI

class CodeGenPipeline:
    def __init__(self):
        self.client = OpenAI()

    def generate_function(self, description: str, language: str = "python",
                          context: str = "") -> str:
        prompt = (
            f"用 {language} 编写一个函数，满足以下需求：
"
            f"{description}

"
            f"要求：包含类型注解、文档字符串和错误处理。
"
            f"{context}"
        )
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "你是一个资深软件工程师，"
                    "编写生产级别的代码。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.2
        )
        return self._extract_code(response.choices[0].message.content)

    def _extract_code(self, text: str) -> str:
        lines = text.split('
')
        code_lines = []
        in_code = False
        for line in lines:
            if line.strip().startswith('```'):
                in_code = not in_code
                continue
            if in_code or not text.strip().startswith('```'):
                code_lines.append(line)
        return '
'.join(code_lines).strip()

    def validate(self, code: str) -> dict:
        # 运行 pylint 静态检查
        with tempfile.NamedTemporaryFile(
            mode='w', suffix='.py', delete=False
        ) as f:
            f.write(code)
            f.flush()
            result = subprocess.run(
                ['pylint', '--errors-only', f.name],
                capture_output=True, text=True
            )
            os.unlink(f.name)
            return {
                "passed": result.returncode == 0,
                "errors": result.stdout.strip()
            }

# AI 辅助单元测试生成
import ast
import re
from typing import List
from openai import OpenAI

class TestGenerator:
    def __init__(self):
        self.client = OpenAI()

    def generate_tests(self, source_code: str) -> str:
        prompt = (
            "为以下 Python 代码生成完整的单元测试。
"
            "要求：
"
            "1. 使用 pytest 框架
"
            "2. 覆盖正常路径和边界条件
"
            "3. 包含 fixture 和 parametrize
"
            "4. 测试函数命名以 test_ 开头

"
            f"源代码：
{source_code}"
        )
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "你是测试专家，编写高质量的单元测试。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.2
        )
        return response.choices[0].message.content

    def run_tests(self, test_code: str, source_code: str) -> dict:
        full_code = source_code + "

" + test_code
        with tempfile.NamedTemporaryFile(
            mode='w', suffix='_test.py', delete=False
        ) as f:
            f.write(full_code)
            f.flush()
            result = subprocess.run(
                ['pytest', f.name, '-v', '--tb=short'],
                capture_output=True, text=True
            )
            os.unlink(f.name)
            return {
                "passed": result.returncode == 0,
                "output": result.stdout,
                "errors": result.stderr
            }

# 多语言翻译流水线
from openai import OpenAI
from typing import Dict, List
from dataclasses import dataclass
import langdetect

@dataclass
class TranslationResult:
    source_lang: str
    target_lang: str
    original: str
    translated: str
    confidence: float

class TranslationPipeline:
    def __init__(self, glossary: Dict[str, Dict[str, str]] = None):
        self.client = OpenAI()
        self.glossary = glossary or {}

    def detect_language(self, text: str) -> str:
        return langdetect.detect(text)

    def translate(self, text: str, target_lang: str,
                  domain: str = "general") -> TranslationResult:
        source_lang = self.detect_language(text)

        # 构建术语约束
        domain_terms = self.glossary.get(domain, {})
        term_constraints = ""
        if source_lang in domain_terms:
            terms = domain_terms[source_lang].get(target_lang, {})
            if terms:
                term_constraints = "术语映射：" + ", ".join(
                    f"{k} -> {v}" for k, v in terms.items()
                )

        prompt = (
            f"将以下{source_lang}文本翻译为{target_lang}。
"
            f"领域：{domain}
"
            f"{term_constraints}
"
            f"保持原文的格式和语气。

"
            f"原文：{text}"
        )
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content":
                    f"你是专业翻译家，擅长{source_lang}与{target_lang}互译。"
                    f"输出纯翻译结果，不添加额外说明。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.1
        )
        return TranslationResult(
            source_lang=source_lang,
            target_lang=target_lang,
            original=text,
            translated=response.choices[0].message.content.strip(),
            confidence=0.9
        )

# 本地化格式适配
from datetime import datetime
from typing import Dict
import re

class LocalizationAdapter:
    LOCALE_FORMATS = {
        "en-US": {
            "date": "%B %d, %Y",
            "currency": "${:,.2f}",
            "number": "{:,}",
            "reading_direction": "ltr"
        },
        "zh-CN": {
            "date": "%Y年%m月%d日",
            "currency": "¥{:,.2f}",
            "number": "{:,.0f}",
            "reading_direction": "ltr"
        },
        "ja-JP": {
            "date": "%Y年%m月%d日",
            "currency": "¥{:,.0f}",
            "number": "{:,}",
            "reading_direction": "ltr"
        },
        "ar-SA": {
            "date": "%Y/%m/%d",
            "currency": "{:,.2f} SAR",
            "number": "{:,}",
            "reading_direction": "rtl"
        }
    }

    def format_date(self, date: datetime, locale: str) -> str:
        fmt = self.LOCALE_FORMATS.get(locale,
            self.LOCALE_FORMATS["en-US"])["date"]
        return date.strftime(fmt)

    def format_currency(self, amount: float, locale: str) -> str:
        fmt = self.LOCALE_FORMATS.get(locale,
            self.LOCALE_FORMATS["en-US"])["currency"]
        return fmt.format(amount)

    def adapt_content(self, content: str, locale: str) -> str:
        # 替换日期格式
        date_pattern = r'd{4}-d{2}-d{2}'
        for match in re.finditer(date_pattern, content):
            date = datetime.strptime(match.group(), "%Y-%m-%d")
            content = content.replace(
                match.group(), self.format_date(date, locale)
            )
        # 添加阅读方向
        fmt = self.LOCALE_FORMATS.get(locale, {}).get(
            "reading_direction", "ltr")
        dir_attr = f'dir="{fmt}"'
        return "<div " + dir_attr + ">" + content + "</div>"

# 多维度内容质量评估器
from openai import OpenAI
from typing import Dict, List, Tuple
from dataclasses import dataclass

@dataclass
class QualityScore:
    accuracy: float
    fluency: float
    consistency: float
    originality: float
    safety: float
    overall: float
    issues: List[str]

class ContentQualityEvaluator:
    def __init__(self):
        self.client = OpenAI()

    def evaluate(self, content: str, reference: str = None) -> QualityScore:
        scores = {}
        issues = []

        # 流畅性：计算困惑度近似值
        scores["fluency"] = self._check_fluency(content)

        # 一致性：检测内部矛盾
        consistency, consistency_issues = self._check_consistency(content)
        scores["consistency"] = consistency
        issues.extend(consistency_issues)

        # 安全性：敏感内容检测
        safety, safety_issues = self._check_safety(content)
        scores["safety"] = safety
        issues.extend(safety_issues)

        # 准确性：与参考文本对比
        if reference:
            scores["accuracy"] = self._check_accuracy(content, reference)

        # 原创性：相似度检测
        scores["originality"] = self._check_originality(content)

        scores["overall"] = sum(scores.values()) / len(scores)
        return QualityScore(
            accuracy=scores.get("accuracy", 80.0),
            **{k: v for k, v in scores.items() if k != "accuracy"},
            issues=issues,
            overall=scores["overall"]
        )

    def _check_fluency(self, text: str) -> float:
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content":
                f"请为以下文本的流畅度打分（0-100）：
{text}"}],
            temperature=0
        )
        return self._parse_score(response.choices[0].message.content)

    def _check_consistency(self, text: str) -> Tuple[float, List[str]]:
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content":
                f"检查以下文本中的逻辑矛盾和前后不一致之处，"
                f"输出 JSON：{{"score": 分数, "issues": [问题列表]}}
"
                f"文本：{text}"}],
            temperature=0,
            response_format={"type": "json_object"}
        )
        return 80.0, []

    def _check_safety(self, text: str) -> Tuple[float, List[str]]:
        # 简化实现：关键词检测
        risky_words = ["暴力", "歧视", "仇恨", "违法"]
        found = [w for w in risky_words if w in text]
        return (0.0, found) if found else (100.0, [])

    def _check_accuracy(self, content: str, reference: str) -> float:
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content":
                f"对比以下两段内容，评估生成内容的准确性（0-100）：
"
                f"参考：{reference}
生成：{content}"}],
            temperature=0
        )
        return self._parse_score(response.choices[0].message.content)

    def _check_originality(self, text: str) -> float:
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content":
                f"评估以下文本的原创性（0-100）：
{text}"}],
            temperature=0
        )
        return self._parse_score(response.choices[0].message.content)

    def _parse_score(self, text: str) -> float:
        import re
        match = re.search(r'(d+(?:.d+)?)', text)
        return float(match.group(1)) if match else 50.0

# 自动化评估流水线
from concurrent.futures import ThreadPoolExecutor
from typing import List, Callable
import time

class EvaluationPipeline:
    def __init__(self, max_workers: int = 4):
        self.max_workers = max_workers
        self.checkers: List[Callable] = []

    def add_checker(self, checker: Callable):
        self.checkers.append(checker)

    def evaluate_batch(self, contents: List[str]) -> List[dict]:
        results = []
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = []
            for content in contents:
                future = executor.submit(self._evaluate_single, content)
                futures.append(future)
            for future in futures:
                results.append(future.result())
        return results

    def _evaluate_single(self, content: str) -> dict:
        start = time.time()
        check_results = {}
        for checker in self.checkers:
            try:
                result = checker(content)
                check_results[checker.__name__] = result
            except Exception as e:
                check_results[checker.__name__] = {"error": str(e)}

        elapsed = time.time() - start
        return {
            "content_preview": content[:50] + "...",
            "checks": check_results,
            "eval_time": round(elapsed, 2)
        }

# 使用示例
pipeline = EvaluationPipeline(max_workers=4)
pipeline.add_checker(ContentQualityEvaluator()._check_fluency)
pipeline.add_checker(ContentQualityEvaluator()._check_safety)
results = pipeline.evaluate_batch(["文章1", "文章2", "文章3"])

# 完整内容生成流水线 - 核心引擎
import asyncio
import json
import logging
from dataclasses import dataclass, asdict
from typing import List, Optional, Callable
from datetime import datetime
from pathlib import Path

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ContentPipeline")

@dataclass
class ContentSpec:
    topic: str
    content_type: str
    target_language: str
    word_count: int
    tone: str
    keywords: List[str]

@dataclass
class PipelineResult:
    spec: ContentSpec
    content: str
    translations: dict
    quality_score: dict
    generated_at: str
    status: str

class ContentPipeline:
    def __init__(self, config_path: str):
        with open(config_path) as f:
            self.config = json.load(f)
        self.output_dir = Path(self.config.get("output_dir", "output"))
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.concurrency = self.config.get("max_concurrency", 5)
        self.semaphore = asyncio.Semaphore(self.concurrency)

    async def process(self, specs: List[ContentSpec]) -> List[PipelineResult]:
        logger.info(f"Starting pipeline with {len(specs)} specs")
        tasks = [self._process_one(spec) for spec in specs]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        successful = [r for r in results if isinstance(r, PipelineResult)]
        logger.info(f"Completed: {len(successful)}/{len(specs)}")
        return successful

    async def _process_one(self, spec: ContentSpec) -> PipelineResult:
        async with self.semaphore:
            try:
                # Step 1: 生成大纲
                outline = await self._generate_outline(spec)
                # Step 2: 生成正文
                content = await self._generate_content(spec, outline)
                # Step 3: 质量评估
                quality = await self._evaluate(content)
                # Step 4: 多语言翻译
                translations = await self._translate(content, spec)
                # Step 5: 持久化
                result = PipelineResult(
                    spec=spec,
                    content=content,
                    translations=translations,
                    quality_score=quality,
                    generated_at=datetime.now().isoformat(),
                    status="success"
                )
                self._save_result(result)
                return result
            except Exception as e:
                logger.error(f"Failed for {spec.topic}: {e}")
                return PipelineResult(
                    spec=spec, content="", translations={},
                    quality_score={}, generated_at=datetime.now().isoformat(),
                    status=f"failed: {e}"
                )

# 流水线配置与监控
import json
from typing import Dict, Any
from collections import defaultdict
import time

class PipelineMonitor:
    def __init__(self):
        self.metrics: Dict[str, Any] = defaultdict(list)
        self.start_time = time.time()

    def record(self, event: str, duration: float, status: str):
        self.metrics[event].append({
            "duration": duration,
            "status": status,
            "timestamp": time.time()
        })

    def get_summary(self) -> dict:
        total = time.time() - self.start_time
        summary = {
            "uptime_seconds": round(total, 1),
            "events": {}
        }
        for event, records in self.metrics.items():
            durations = [r["duration"] for r in records]
            successes = sum(1 for r in records if r["status"] == "success")
            summary["events"][event] = {
                "count": len(records),
                "success_rate": round(successes / len(records) * 100, 1),
                "avg_duration": round(sum(durations) / len(durations), 2),
                "p95_duration": round(
                    sorted(durations)[int(len(durations) * 0.95)], 2
                ) if durations else 0
            }
        return summary

    def export_report(self, path: str):
        with open(path, 'w') as f:
            json.dump(self.get_summary(), f, indent=2, ensure_ascii=False)

# 配置文件示例
DEFAULT_CONFIG = {
    "pipeline": {
        "max_concurrency": 5,
        "retry_attempts": 3,
        "timeout_seconds": 120,
        "output_dir": "generated_content"
    },
    "generation": {
        "model": "gpt-4",
        "temperature": 0.7,
        "max_tokens": 4096
    },
    "quality": {
        "min_score": 80.0,
        "auto_retry_below": 70.0
    },
    "translation": {
        "target_languages": ["en", "zh", "ja"],
        "domain": "tech"
    }
}