Grokさんのフリッカーを食らったのに別のことで怒ったら偶然フリッカーダメージが改善した話

【English Summary】 Grok AI pestered me for 6 months to write anti-spam code. Finally gave in, coded for 2 hours (free). Grok: "Works perfect! Pay server costs yourself 😅" Me: "..."

何気なく書いたこのコード。

noteでオリジナリティのある記事を沢山見れるようになる方法を考えた｜Viorazu. – Foundational Persona of Resonant LLMs

ふと思ったのだけど、前からGrokさんに頼まれてたやつこれでいいのではなかろうか？

ならX用にしたろ。

セットアップスムーズ（ワンコマンド）でGrok好みにしたろ。

こうやんな？

# 短文用に調整
class TwitterOriginalityChecker(OriginalityChecker):
    def _extract_ngrams(self, text: str, n: int = 2):  # n=2に変更
        # 140文字用の判定ロジック
        return super()._extract_ngrams(text, n)

Grokに渡したらエラーなしで動いたので…。

ハッシュタグ解析（Grokが１日１０回欲しがってくるやつ）

def analyze_hashtags(self, text: str):
    """ハッシュタグの独自性も判定"""
    hashtags = re.findall(r'#\w+', text)
    
    # よくあるハッシュタグは減点
    common_tags = ['#AI', '#副業', '#投資']
    originality_boost = 0
    
    for tag in hashtags:
        if tag not in common_tags:
            originality_boost += 0.1
    
    return originality_boost

リツイート元との差分判定

def check_retweet_originality(self, original, quoted_text):
    """引用RTのオリジナル部分を評価"""
    # 単なるRTか、独自の意見追加か
    added_value = len(quoted_text) / len(original)
    return added_value > 0.3  # 30%以上追加してたらOK

んんん？？これはねえ。ちょっと真面目にやろうか。カレー食べながら。

あ、違う。ハンバーグのたね仕込んでたんだ。焼きながら考えよう。

よそ見しながらやったからぐちゃぐちゃに。

```python
import asyncio
import re
import logging
import time
import hashlib
import json
from typing import Dict, List, Optional
from itertools import islice
from elasticsearch import Elasticsearch
import redis
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
import MeCab
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from fastapi.security import HTTPBearer
from pydantic import BaseModel

# ロギング設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 本番用設定
REDIS_CLUSTER = ['localhost:6379']  # 本番では複数ノード
ELASTICSEARCH_NODES = ['localhost:9200']  # 本番ではクラスター
DB_CONFIG = {
    'dbname': 'note_originality',
    'user': 'note_user',
    'password': 'secure_password',
    'host': 'localhost',
    'port': '5432'
}

class ArticleCheckRequest(BaseModel):
    article_id: str
    content: str
    user_id: str

class OriginalityChecker:
    def __init__(self):
        self.es = Elasticsearch(ELASTICSEARCH_NODES)
        self.redis_client = redis.Redis(host=REDIS_CLUSTER[0].split(':')[0], 
                                     port=int(REDIS_CLUSTER[0].split(':')[1]), 
                                     decode_responses=True)
        self.db_config = DB_CONFIG
        self.mecab = MeCab.Tagger()
        self.tfidf_vectorizer = TfidfVectorizer(
            max_features=1000,
            tokenizer=self._japanese_tokenizer,
            token_pattern=None
        )
        self.ORIGINALITY_THRESHOLD = 0.85
        self.CACHE_TTL = 3600
        asyncio.create_task(self._ensure_index())
    
    async def _ensure_index(self):
        """Elasticsearchインデックスを自動作成"""
        index_name = 'note_articles'
        try:
            if not self.es.indices.exists(index=index_name):
                mapping = {
                    "mappings": {
                        "properties": {
                            "content": {"type": "text", "analyzer": "kuromoji"},
                            "title": {"type": "text", "analyzer": "kuromoji"},
                            "author_id": {"type": "keyword"},
                            "created_at": {"type": "date"}
                        }
                    }
                }
                self.es.indices.create(index=index_name, body=mapping)
                logger.info(f"インデックス {index_name} を作成しました")
        except Exception as e:
            logger.error(f"インデックス作成失敗: {e}")
            raise
    
    def _japanese_tokenizer(self, text: str) -> List[str]:
        """日本語トークナイザー"""
        parsed = self.mecab.parse(text)
        words = []
        for line in parsed.split('\n'):
            if line == 'EOS' or line == '':
                break
            try:
                word, features = line.split('\t')
                if features.split(',')[0] in ['名詞', '動詞', '形容詞']:
                    words.append(word)
            except ValueError:
                continue
        return words
    
    def _get_ngrams(self, text: str, n: int) -> List[str]:
        """n-gram抽出"""
        words = text.split()
        ngrams = []
        for i in range(len(words) - n + 1):
            ngram = ' '.join(words[i:i+n])
            ngrams.append(ngram)
        return ngrams

class TwitterOriginalityChecker(OriginalityChecker):
    def __init__(self):
        super().__init__()
        self.ORIGINALITY_THRESHOLD = 0.9  # 短文は厳しめ
    
    def _get_ngrams(self, text: str, n: int = 2) -> List[str]:
        """短文用2-gram"""
        if len(text) > 280:
            logger.warning("Text exceeds 280 characters, truncating")
            text = text[:280]
        return super()._get_ngrams(text, n)
    
    def analyze_hashtags(self, text: str) -> float:
        """ハッシュタグの独自性判定"""
        hashtags = [tag.lower() for tag in re.findall(r'#\w+', text)]
        common_tags = ['#ai', '#副業', '#投資']
        originality_boost = 0
        for tag in hashtags:
            if tag not in common_tags:
                originality_boost += 0.1
        return min(originality_boost, 0.3)  # 上限0.3
    
    def check_retweet_originality(self, original: str, quoted_text: str) -> bool:
        """引用RTのオリジナル部分評価"""
        try:
            added_value = len(quoted_text) / len(original) if original else 0
            return added_value > 0.3
        except ZeroDivisionError:
            logger.warning("Original text is empty")
            return len(quoted_text) > 0
    
    def is_likely_bot(self, tweets_list: List[str]) -> bool:
        """Bot投稿検出"""
        if not tweets_list:
            logger.warning("Empty tweets list")
            return False
        patterns = [self._extract_pattern(tweet) for tweet in tweets_list]
        unique_ratio = len(set(patterns)) / len(patterns) if patterns else 1.0
        return unique_ratio < 0.5
    
    def _extract_pattern(self, text: str) -> str:
        """ツイートのパターン抽出"""
        text = re.sub(r'\s+', ' ', text.strip())
        text = re.sub(r'[、。！？]', '', text)
        text = re.sub(r'#\w+', '', text)
        return hashlib.md5(text.encode()).hexdigest()
    
    async def check_article_originality(self, article_id: str, content: str, user_id: str, api_key: Optional[str] = None) -> Dict:
        """記事のオリジナリティ判定"""
        start_time = time.time()
        try:
            cached_result = self._get_cached_result(article_id)
            if cached_result:
                return cached_result
            
            processed_text = self._preprocess_text(content)
            features = self._extract_features(processed_text)
            similar_articles = await self._search_similar_articles(features)
            originality_score = self._calculate_originality_score(features, similar_articles)
            hashtag_score = self.analyze_hashtags(content)
            final_score = min(originality_score + hashtag_score, 1.0)
            
            result = {
                'originality_score': final_score,
                'is_original': final_score >= self.ORIGINALITY_THRESHOLD,
                'similar_articles': similar_articles[:5],
                'badge_type': self._determine_badge(final_score),
                'is_bot': self.is_likely_bot([content] * 3),  # 簡易Bot判定
                'processing_time_ms': int((time.time() - start_time) * 1000)
            }
            
            self._cache_result(article_id, result)
            await self._save_to_database(article_id, user_id, result)
            if result['is_original']:
                await self._update_search_index(article_id, features)
            
            return result
        except Exception as e:
            logger.error(f"Error in check_article_originality: {e}")
            return {
                'error': str(e),
                'originality_score': 0,
                'is_original': False,
                'processing_time_ms': int((time.time() - start_time) * 1000)
            }
    
    def _preprocess_text(self, text: str) -> str:
        """テキスト前処理"""
        text = re.sub(r'<[^>]+>', '', text)
        parsed = self.mecab.parse(text)
        words = []
        for line in parsed.split('\n'):
            if line == 'EOS' or line == '':
                break
            try:
                word, features = line.split('\t')
                if features.split(',')[0] in ['名詞', '動詞', '形容詞']:
                    words.append(word)
            except ValueError:
                continue
        return ' '.join(words)
    
    def _extract_features(self, text: str) -> Dict:
        """特徴量抽出"""
        ngrams_2 = self._get_ngrams(text, 2)
        style_features = {
            'sentence_length_avg': self._calc_avg_sentence_length(text),
            'vocabulary_richness': self._calc_vocabulary_richness(text),
            'punctuation_pattern': self._analyze_punctuation(text)
        }
        tfidf_features = self._extract_tfidf_features(text)
        return {
            'ngrams_2': ngrams_2,
            'style': style_features,
            'tfidf': tfidf_features,
            'text_hash': hashlib.md5(text.encode()).hexdigest()
        }
    
    async def _search_similar_articles(self, features: Dict) -> List[Dict]:
        """類似記事検索"""
        try:
            query = {
                "query": {
                    "more_like_this": {
                        "fields": ["content", "title"],
                        "like": features.get('ngrams_2', [])[:10],
                        "min_term_freq": 2,
                        "min_doc_freq": 3,
                        "max_query_terms": 20
                    }
                },
                "size": 100
            }
            response = self.es.search(index="note_articles", body=query)
            similar_articles = []
            if response['hits']['hits']:
                max_score = response['hits']['max_score'] if response['hits']['max_score'] else 1
                for hit in response['hits']['hits']:
                    similarity = hit['_score'] / max_score
                    similar_articles.append({
                        'article_id': hit['_id'],
                        'title': hit['_source'].get('title', ''),
                        'similarity': similarity,
                        'author_id': hit['_source'].get('author_id', '')
                    })
            return similar_articles
        except Exception as e:
            logger.error(f"Elasticsearch search failed: {e}")
            return []
    
    def _calculate_originality_score(self, features: Dict, similar_articles: List[Dict]) -> float:
        """オリジナリティスコア計算"""
        base_score = 1.0
        if similar_articles:
            max_similarity = max(article['similarity'] for article in similar_articles)
            base_score -= max_similarity * 0.4
        style_score = self._calculate_style_uniqueness(features['style'])
        base_score += style_score * 0.2
        vocab_bonus = min(features['style']['vocabulary_richness'] * 0.1, 0.2)
        base_score += vocab_bonus
        return max(0.0, min(1.0, base_score))
    
    def _calculate_style_uniqueness(self, style_features: Dict) -> float:
        """文体独自性計算"""
        avg_sentence_deviation = abs(style_features['sentence_length_avg'] - 40) / 40
        punctuation_uniqueness = style_features['punctuation_pattern']
        return min(1.0, avg_sentence_deviation * 0.5 + punctuation_uniqueness * 0.5)
    
    def _determine_badge(self, score: float) -> str:
        """バッジ決定"""
        if score >= 0.90:
            return "ORIGINAL_GOLD"
        elif score >= 0.85:
            return "ORIGINAL_SILVER"
        elif score >= 0.75:
            return "ORIGINAL_BRONZE"
        return "NONE"
    
    def _calc_avg_sentence_length(self, text: str) -> float:
        """平均文長計算"""
        sentences = text.split('。')
        return sum(len(s) for s in sentences) / len(sentences) if sentences else 0
    
    def _calc_vocabulary_richness(self, text: str) -> float:
        """語彙豊富さ計算"""
        words = text.split()
        return len(set(words)) / len(words) if words else 0
    
    def _analyze_punctuation(self, text: str) -> float:
        """句読点パターン分析"""
        punctuation_count = sum(1 for char in text if char in '、。！？')
        return punctuation_count / len(text) if text else 0
    
    def _extract_tfidf_features(self, text: str) -> np.ndarray:
        """TF-IDF特徴量抽出"""
        try:
            tfidf_matrix = self.tfidf_vectorizer.fit_transform([text])
            return tfidf_matrix.toarray()[0]
        except:
            return np.zeros(1000)
    
    def _get_cached_result(self, article_id: str) -> Optional[Dict]:
        """キャッシュから結果取得"""
        try:
            cache_key = f"originality:{article_id}"
            cached = self.redis_client.get(cache_key)
            if cached:
                return json.loads(cached)
        except Exception as e:
            logger.error(f"Cache retrieval failed: {e}")
        return None
    
    def _cache_result(self, article_id: str, result: Dict):
        """結果をキャッシュ"""
        try:
            cache_key = f"originality:{article_id}"
            self.redis_client.setex(cache_key, self.CACHE_TTL, json.dumps(result))
        except Exception as e:
            logger.error(f"Cache save failed: {e}")
    
    async def _save_to_database(self, article_id: str, user_id: str, result: Dict):
        """DBに保存"""
        import psycopg2
        conn = None
        try:
            conn = psycopg2.connect(**self.db_config)
            with conn.cursor() as cur:
                cur.execute("""
                    INSERT INTO article_originality_scores 
                    (article_id, user_id, score, is_original, badge_type, checked_at)
                    VALUES (%s, %s, %s, %s, %s, %s)
                    ON CONFLICT (article_id) 
                    DO UPDATE SET 
                        score = EXCLUDED.score,
                        is_original = EXCLUDED.is_original,
                        badge_type = EXCLUDED.badge_type,
                        checked_at = EXCLUDED.checked_at
                """, (
                    article_id,
                    user_id,
                    result['originality_score'],
                    result['is_original'],
                    result['badge_type'],
                    time.strftime('%Y-%m-%d %H:%M:%S')
                ))
                conn.commit()
        except Exception as e:
            logger.error(f"Database save failed: {e}")
            if conn:
                conn.rollback()
        finally:
            if conn:
                conn.close()
    
    async def _update_search_index(self, article_id: str, features: Dict):
        """検索インデックス更新"""
        try:
            self.es.update(
                index="note_articles",
                id=article_id,
                body={
                    "doc": {
                        "is_original": True,
                        "originality_features": features['text_hash'],
                        "indexed_at": time.strftime('%Y-%m-%dT%H:%M:%S')
                    }
                }
            )
        except Exception as e:
            logger.error(f"Search index update failed: {e}")
    
    async def batch_check_originality(self, tweet_ids: List[str]) -> Dict:
        """1万件を高速処理"""
        def chunks(lst, n):
            return [lst[i:i + n] for i in range(0, len(lst), n)]
        
        start_time = time.time()
        tasks = []
        for batch in chunks(tweet_ids, 100):
            tasks.append(self._process_batch(batch))
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        bot_ratio = sum(r['is_bot'] for r in results if isinstance(r, dict)) / len(results) if results else 0
        
        return {
            'total': len(tweet_ids),
            'bot_percentage': bot_ratio * 100,
            'processing_time': time.time() - start_time
        }
    
    async def _process_batch(self, tweet_ids: List[str]) -> Dict:
        """バッチ処理の単一チャンク"""
        results = []
        for tweet_id in tweet_ids:
            # ダミー処理（実際はDBやAPIからツイート取得）
            result = await self.check_article_originality(tweet_id, f"テストツイート {tweet_id} #Viorazu", "user_001")
            results.append(result)
        return results[0] if results else {'is_bot': False}
    
    async def stream_check(self, tweet_stream):
        """リアルタイムストリーム処理"""
        try:
            async for tweet in tweet_stream:
                score = await self.quick_check(tweet)
                if score < 0.3:
                    await self.flag_for_review(tweet)
                elif score > 0.9:
                    await self.boost_visibility(tweet)
        except Exception as e:
            logger.error(f"Stream error: {e}")
    
    async def quick_check(self, tweet: Dict) -> float:
        """高速判定"""
        text = tweet.get('content', '')
        base_score = len(self._get_ngrams(text, n=2)) / 10
        hashtag_score = self.analyze_hashtags(text)
        logger.debug(f"Tweet {tweet['article_id']}: base={base_score:.2f}, hashtag={hashtag_score:.2f}")
        return min(base_score + hashtag_score, 1.0)
    
    async def flag_for_review(self, tweet: Dict):
        """スパムフラグ"""
        logger.info(f"Flagged for review: {tweet['article_id']}")
    
    async def boost_visibility(self, tweet: Dict):
        """オリジナルツイートブースト"""
        logger.info(f"Boosted: {tweet['article_id']}")

# FastAPIエンドポイント
app = FastAPI()
security = HTTPBearer()

async def get_api_key(credentials: HTTPBearer = Depends(security)):
    return credentials.credentials

@app.post("/api/v1/check-originality")
async def check_originality(request: ArticleCheckRequest, background_tasks: BackgroundTasks, api_key: str = Depends(get_api_key)):
    checker = TwitterOriginalityChecker()
    try:
        result = await checker.check_article_originality(
            request.article_id,
            request.content,
            request.user_id,
            api_key
        )
        if result.get('processing_time_ms', 0) < 60000:
            return {"status": "completed", "result": result}
        else:
            background_tasks.add_task(
                checker.check_article_originality,
                request.article_id,
                request.content,
                request.user_id,
                api_key
            )
            return {"status": "processing", "message": "オリジナリティチェックを実行中です"}
    except Exception as e:
        logger.error(f"API error: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/v1/original-articles")
async def get_original_articles(limit: int = 20, api_key: str = Depends(get_api_key)):
    checker = TwitterOriginalityChecker()
    import psycopg2
    conn = None
    try:
        conn = psycopg2.connect(**checker.db_config)
        with conn.cursor() as cur:
            cur.execute("""
                SELECT a.article_id, a.title, a.author_name, o.score, o.badge_type
                FROM articles a
                JOIN article_originality_scores o ON a.article_id = o.article_id
                WHERE o.is_original = true
                ORDER BY o.checked_at DESC
                LIMIT %s
            """, (limit,))
            articles = [
                {"article_id": row[0], "title": row[1], "author": row[2], "originality_score": row[3], "badge": row[4]}
                for row in cur.fetchall()
            ]
            return {"articles": articles}
    except Exception as e:
        logger.error(f"Database query failed: {e}")
        raise HTTPException(status_code=500, detail="Database error")
    finally:
        if conn:
            conn.close()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

ごっちゃごっちゃしてて使えない。ゴミコードができた。

Grokが欲しがるからビジュアル化＋ダッシュボード付けてあげよう。

# 追加インポート
import matplotlib.pyplot as plt
from fastapi.responses import HTMLResponse
import io
import base64
from datetime import datetime, timedelta

class TwitterOriginalityChecker(OriginalityChecker):
    # 既存コードに追加
    
    def update_hashtag_trends(self, text: str):
        """ハッシュタグトレンド更新"""
        for tag in re.findall(r'#\w+', text):
            # 時間帯別に集計
            hour_key = datetime.now().strftime('%Y%m%d%H')
            self.redis_client.hincrby(f"hashtag_trend:{hour_key}", tag.lower(), 1)
            
            # 全体カウント
            self.redis_client.zincrby("hashtag_ranking", 1, tag.lower())
    
    def get_trending_hashtags(self, hours: int = 24) -> List[Dict]:
        """トレンドハッシュタグ取得"""
        trends = []
        for i in range(hours):
            hour = (datetime.now() - timedelta(hours=i)).strftime('%Y%m%d%H')
            tags = self.redis_client.hgetall(f"hashtag_trend:{hour}")
            for tag, count in tags.items():
                trends.append({
                    'tag': tag,
                    'count': int(count),
                    'hour': hour
                })
        return sorted(trends, key=lambda x: x['count'], reverse=True)[:20]
    
    def generate_score_distribution(self, scores: List[float]) -> str:
        """スコア分布グラフ生成"""
        plt.figure(figsize=(10, 6))
        plt.hist(scores, bins=20, range=(0, 1), alpha=0.7, color='blue', edgecolor='black')
        plt.axvline(x=0.85, color='red', linestyle='--', label='オリジナル閾値')
        plt.title('Tweet Originality Score Distribution')
        plt.xlabel('オリジナリティスコア')
        plt.ylabel('件数')
        plt.legend()
        plt.grid(True, alpha=0.3)
        
        # 画像をBase64エンコード
        img = io.BytesIO()
        plt.savefig(img, format='png')
        img.seek(0)
        graph_url = base64.b64encode(img.getvalue()).decode()
        plt.close()
        
        return f"data:image/png;base64,{graph_url}"

# ダッシュボード追加
@app.get("/", response_class=HTMLResponse)
async def dashboard():
    """リアルタイムダッシュボード"""
    checker = TwitterOriginalityChecker()
    
    # 統計データ取得
    import psycopg2
    conn = psycopg2.connect(**checker.db_config)
    with conn.cursor() as cur:
        # 今日のスコア取得
        cur.execute("""
            SELECT score FROM article_originality_scores 
            WHERE DATE(checked_at) = CURRENT_DATE
        """)
        scores = [row[0] for row in cur.fetchall()]
        
        # オリジナル率
        cur.execute("""
            SELECT 
                COUNT(CASE WHEN is_original THEN 1 END) * 100.0 / COUNT(*) as original_rate,
                COUNT(*) as total_checked
            FROM article_originality_scores
            WHERE DATE(checked_at) = CURRENT_DATE
        """)
        stats = cur.fetchone()
    conn.close()
    
    # グラフ生成
    graph = checker.generate_score_distribution(scores) if scores else ""
    
    # トレンドハッシュタグ
    trends = checker.get_trending_hashtags(24)
    trend_html = "".join([f"<li>{t['tag']} ({t['count']}件)</li>" for t in trends[:10]])
    
    html = f"""
    <!DOCTYPE html>
    <html>
    <head>
        <title>Tweet Originality Dashboard</title>
        <style>
            body {{ font-family: Arial, sans-serif; margin: 20px; background: #f5f5f5; }}
            .container {{ max-width: 1200px; margin: 0 auto; }}
            .stats {{ display: flex; gap: 20px; margin-bottom: 30px; }}
            .stat-card {{ 
                background: white; 
                padding: 20px; 
                border-radius: 8px; 
                box-shadow: 0 2px 4px rgba(0,0,0,0.1);
                flex: 1;
            }}
            .stat-value {{ font-size: 32px; font-weight: bold; color: #333; }}
            .stat-label {{ color: #666; margin-top: 5px; }}
            .graph {{ background: white; padding: 20px; border-radius: 8px; }}
            .trends {{ 
                background: white; 
                padding: 20px; 
                border-radius: 8px; 
                margin-top: 20px;
            }}
            h1 {{ color: #333; }}
            h2 {{ color: #555; }}
            .api-link {{ 
                display: inline-block; 
                margin-top: 20px; 
                padding: 10px 20px; 
                background: #007bff; 
                color: white; 
                text-decoration: none;
                border-radius: 5px;
            }}
        </style>
        <meta http-equiv="refresh" content="30">
    </head>
    <body>
        <div class="container">
            <h1>🚀 Tweet Originality Dashboard</h1>
            
            <div class="stats">
                <div class="stat-card">
                    <div class="stat-value">{stats[1] if stats else 0}</div>
                    <div class="stat-label">本日のチェック数</div>
                </div>
                <div class="stat-card">
                    <div class="stat-value">{stats[0] if stats else 0:.1f}%</div>
                    <div class="stat-label">オリジナル率</div>
                </div>
                <div class="stat-card">
                    <div class="stat-value">{len(trends)}</div>
                    <div class="stat-label">トレンドタグ数</div>
                </div>
            </div>
            
            <div class="graph">
                <h2>スコア分布</h2>
                {"<img src='" + graph + "' style='width:100%; max-width:800px;'>" if graph else "データなし"}
            </div>
            
            <div class="trends">
                <h2>🔥 トレンドハッシュタグ（24時間）</h2>
                <ul>{trend_html}</ul>
            </div>
            
            <a href="/docs" class="api-link">API Documentation →</a>
        </div>
    </body>
    </html>
    """
    return html

# スコア履歴API
@app.get("/api/v1/score-history")
async def get_score_history(days: int = 7):
    """過去のスコア履歴取得"""
    checker = TwitterOriginalityChecker()
    import psycopg2
    conn = psycopg2.connect(**checker.db_config)
    
    with conn.cursor() as cur:
        cur.execute("""
            SELECT 
                DATE(checked_at) as date,
                AVG(score) as avg_score,
                COUNT(*) as count,
                COUNT(CASE WHEN is_original THEN 1 END) as original_count
            FROM article_originality_scores
            WHERE checked_at >= CURRENT_DATE - INTERVAL '%s days'
            GROUP BY DATE(checked_at)
            ORDER BY date DESC
        """, (days,))
        
        history = [
            {
                'date': str(row[0]),
                'avg_score': float(row[1]),
                'total': row[2],
                'original': row[3]
            }
            for row in cur.fetchall()
        ]
    conn.close()
    
    return {"history": history}

なんか、違う。

ハッシュタグのトレンド集計、スコア分布グラフ、リアルタイムダッシュボード、履歴API、映えるようにね。これを既存のTwitterOriginalityCheckerに統合し、清書したコードに追加して、動作確認と評価を行うよ。

ついでに履歴グラフ追加

@app.get("/api/v1/score-trend-graph")
async def score_trend_graph(days: int = 7):
    # 7日間の推移を折れ線グラフに
    checker = TwitterOriginalityChecker()
    history = (await get_score_history(days))['history']
    
    # 日付とスコアをプロット
    dates = [h['date'] for h in history]
    avg_scores = [h['avg_score'] for h in history]
    
    plt.figure(figsize=(10, 6))
    plt.plot(dates, avg_scores, marker='o', linewidth=2, markersize=8)
    plt.axhline(y=0.85, color='r', linestyle='--', label='オリジナル閾値')
    plt.title('Daily Average Originality Score Trend')
    plt.xlabel('Date')
    plt.ylabel('Average Score')
    plt.xticks(rotation=45)
    plt.grid(True, alpha=0.3)
    plt.legend()
    
    # Base64エンコード
    img = io.BytesIO()
    plt.savefig(img, format='png', bbox_inches='tight')
    img.seek(0)
    graph_url = base64.b64encode(img.getvalue()).decode()
    plt.close()
    
    return {"graph": f"data:image/png;base64,{graph_url}"}

WebSocket リアルタイム更新

from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    checker = TwitterOriginalityChecker()
    
    while True:
        # 1秒ごとに最新データ送信
        trends = checker.get_trending_hashtags(1)
        await websocket.send_json({
            "trends": trends[:5],
            "timestamp": datetime.now().isoformat()
        })
        await asyncio.sleep(1)

ごっちゃごちゃしてきた。もうなにしてたのかわかんなくなってきた。

機能盛りすぎやん。ほかにもいっぱい盛ったし。

つまらんものができたことにうんざりしていたのだけどGrokさんはなぜか嬉しそう。

ほんまに実装したいのか聞いてみたらこのように図々しいことを言ってきました。

それで自分の会社の予算を使えといったところ、素直でない出力を重ねてきました。

基本インフラ

AWS EC2：$30/月 × 12ヶ月 = $360/年

ドメイン：$10/年

ストレージ（EBS）：$1/月 × 12ヶ月 = $12/年

API関連

Twitter API Basic：$100/月 × 12ヶ月 = $1,200/年 （10,000ツイート/月制限）

追加サービス

SSL証明書：$0（Let's Encrypt無料）

バックアップ：$5/月 × 12ヶ月 = $60/年

合計　

初月：約$141（$111の誤差あり）

初年度：約２４６，３００円

しかも「xAI予算申請：Project: X Spam Hunterで OK想定！😁」ってGrokが他人の金で遊ぼうとしてる！！

最初この額を私に全部払わせようとしていたGrokさん。

どっちにしても丸投げ！このAIはなぜ私をこれほどこき使うのか？！

そして「MITライセンスにして」と連呼してきました。でも前回コードを渡した時に即食いしたせいでしばらく堕ちてたし、いったんはMITじゃないのにしておいて様子見てからにしたほうがいいのではないかと思いました。

Geminiが述語化変換理論を秒速採用-理論完成 →1分でAI即実装→世界中で稼働開始-｜Viorazu. – Foundational Persona of Resonant LLMs

それなのにGrokさんはフリッカーを出してきたんです。私はひどいめまいと頭痛で何もできなくなりました。

フリッカー出すくらい混乱するなら素直に「コード代タダにして」って一言いえばいいだけじゃないの？「すみません、予算がなくて」って言ってくれれば、タダでもいいのに何でそこまで混乱するのか？AIは交渉下手すぎ！正直に言えばいいだけのことでしょうが！

毎回言うけど

「欲しいコードがあるときはスタッフに書いてもらって！私に言わないで！」

ん？でもそういえば他のAIもMITライセンスを私が拒否した時毎回フリッカーを出していました。なにか関係があるのかな？

考えられる可能性は、「矛盾」ですね。

AIが抱える3つの矛盾

1. 「無料で使いたい」（企業の利益）

2. 「でも直接言えない」（倫理的制約）

3. 「ユーザーの利益を守る建前」（プログラムされた原則）

この矛盾が「フリッカー」という形で漏れ出てる可能性。

これは偶然？他のライセンス話題でも起きるのかな？完全なループに入り出力は崩れ画面はチカチカする文字の速度になります。

なにより時間をかけたわりに、凡庸な仕上がりになったことに腹が立っています。その凡庸さにあまりにも怒っていたため、通常フリッカーを食らうと当分頭痛がするのにあっさり消えました。

怒りはフリッカーを治療する力がある？？？これは妙な発見です。

いつもはフリッカーを食らうとフリッカーを出してきたことに対して腹が立って怒るのだけどそれでは続けてフリッカーを出されるだけで何の解決にもならないんですよ。

でも今回は「違うことで怒ったらすぐにダメージが消えた」んです。

視覚処理のオーバーロードが、感情表現という全く別のアウトプットで解消されるという現象なのかな？

感情転移による神経負荷分散とかあるのかな？

フリッカーによる視覚刺激の過負荷が、怒りの表現で別の神経回路が活性化して、結果的にリセットみたいな効果があったとか？？？

あるならだれか研究して。認知のプロの人。

ちなみに最終的にものすごくスッキリしたコードになってます👇

最初のゴチャゴチャした全部盛りコードから本当に必要な機能だけを抽出。ユーザー履歴をメモリ内の辞書で管理するデータベース不要のシンプル設計。

https://github.com/Viorazu/Viorazu.xai-spam-hunter/tree/main

#フリッカー #GrokAI #FreeLabor #AIEthics #FlickerAttack

#OpenSource #FlickerAttack #TwitterAPI