NarrativeAnalysis — 大数据电影叙事分析

# ── 情感分析核心 ──
def analyze_script(filepath, movie_id, movie_name, word_dict):
    """分析单部剧本，返回结构化的情感 + 角色数据"""
    with open(filepath, 'r', encoding='utf-8') as f:
        text = f.read()

    lines = text.split('\n')
    scenes = []
    current_scene = {'scene_id': 0, 'name': 'UNKNOWN',
                     'lines': [], 'characters': set()}
    scene_counter = 0
    all_characters = set()
    char_scene_map = defaultdict(set)

    for line in lines:
        stripped = line.strip()
        # 识别场景标记 [SCENE]
        if stripped.startswith('[SCENE]'):
            if current_scene['lines']:
                scenes.append(current_scene)
            scene_counter += 1
            current_scene = {
                'scene_id': scene_counter,
                'name': stripped.replace('[SCENE] ', '').strip(),
                'lines': [], 'characters': set()
            }
        # 识别角色名（全大写缩写）
        elif stripped and re.match(r'^[A-Z][A-Z\s\.]{1,30}$', stripped):
            char_name = stripped.strip()
            all_characters.add(char_name)
            current_scene['characters'].add(char_name)
            char_scene_map[char_name].add(current_scene['scene_id'])
        else:
            current_scene['lines'].append(stripped)

    if current_scene['lines']:
        scenes.append(current_scene)

    # 无场景标记时按窗口分割
    if len(scenes) <= 1:
        scenes = []
        window_size = 50
        for i in range(0, len(lines), window_size):
            ...  # windowed fallback

    # ── 情感分析 ──
    emotion_results = []
    for scene in scenes:
        pos_count = 0
        neg_count = 0
        total_words = 0
        for line in scene['lines']:
            words = line.lower().split()
            total_words += len(words)
            for w in words:
                w_clean = re.sub(r'[^a-z]', '', w)
                if w_clean in word_dict:
                    p, n = word_dict[w_clean]
                    pos_count += p
                    neg_count += n

        net_emotion = round((pos_count - neg_count) / max(total_words, 1), 6)
        emotion_results.append({
            'scene': scene['scene_id'],
            'name': scene['name'],
            'pos': pos_count, 'neg': neg_count,
            'net': net_emotion
        })

    # ── 角色共现 ──
    cooccur = defaultdict(int)
    for scene in scenes:
        chars = list(scene['characters'])
        for i in range(len(chars)):
            for j in range(i + 1, len(chars)):
                pair = tuple(sorted([chars[i], chars[j]]))
                cooccur[pair] += 1

    return {
        'id': movie_id, 'name': movie_name,
        'total_scenes': len(scenes),
        'total_characters': len(all_characters),
        'scenes': emotion_results,
        'cooccurrences': [
            {'charA': a, 'charB': b, 'count': c}
            for (a, b), c in sorted(cooccur.items(), key=lambda x: -x[1])[:30]
        ]
    }


# ── HBase 写入 ──
def write_to_hbase(movies_data):
    """将分析结果写入 HBase（3 张表）"""
    connection = happybase.Connection('localhost', 9090)
    table_emotion = connection.table('movie_emotion')
    table_network = connection.table('character_network')
    table_pattern = connection.table('narrative_pattern')

    for movie in movies_data:
        mid = movie['id']
        # movie_emotion: 场景情感序列
        table_emotion.put(mid, {
            'emotion:scenes': json.dumps(movie['scenes']),
            'meta:name': movie['name'],
            'meta:scenes': str(movie['total_scenes']),
            'meta:characters': str(movie['total_characters']),
        })
        # character_network: 角色共现关系
        for c in movie['cooccurrences']:
            pair_key = f"{c['charA']}#{c['charB']}"
            table_network.put(mid, {f"cooccur:{pair_key}": str(c['count'])})
        # narrative_pattern: 叙事模式指纹
        emotion_series = [s['net'] for s in movie['scenes']]
        mean_emo = sum(emotion_series) / max(len(emotion_series), 1)
        std_emo = (sum((e - mean_emo)**2 for e in emotion_series) / max(len(emotion_series), 1))**0.5
        pattern = {
            'acts': len(movie['scenes']) // 10 + 1,
            'emotion_mean': round(mean_emo, 6),
            'emotion_std': round(std_emo, 6),
            'scene_count': movie['total_scenes'],
            'char_count': movie['total_characters'],
        }
        table_pattern.put(mid, {
            'features:fingerprint': json.dumps(pattern),
            'acts:count': str(pattern['acts']),
        })
    connection.close()

@app.route('/movies', methods=['GET'])
def list_movies():
    """获取所有电影基本信息列表"""
    data = load_all_movies()
    movies = [{
        'id': m['id'], 'name': m['name'],
        'total_scenes': m['total_scenes'],
        'total_characters': m['total_characters'],
    } for m in data.get('movies', [])]
    return jsonify({'count': len(movies), 'movies': movies})


@app.route('/movies/<movie_id>/emotion', methods=['GET'])
def get_emotion_curve(movie_id):
    """获取电影情感曲线（含转折点检测）"""
    movie = get_movie_by_id(movie_id)
    if movie is None:
        return jsonify({'error': f'Movie not found: {movie_id}'}), 404

    scenes = movie.get('scenes', [])
    emotion_curve = [{
        'scene': s['scene'], 'name': s['name'],
        'positive': s['pos'], 'negative': s['neg'],
        'net_emotion': s['net'],
    } for s in scenes]

    # 检测叙事转折点（ECR 简化版）
    turning_points = []
    for i in range(1, len(scenes) - 1):
        prev_net = scenes[i - 1]['net']
        curr_net = scenes[i]['net']
        next_net = scenes[i + 1]['net']
        # 局部极值 = 情感趋势转折
        if (curr_net > prev_net and curr_net > next_net) or \
           (curr_net < prev_net and curr_net < next_net):
            turning_points.append({
                'scene': scenes[i]['scene'],
                'net_emotion': curr_net,
                'type': 'peak' if curr_net > prev_net else 'valley'
            })

    return jsonify({
        'movie_id': movie_id,
        'emotion_curve': emotion_curve,
        'turning_points': turning_points,
        'total_scenes': len(scenes),
    })