statistics_model2025/covers_analyser.py

import cv2
import numpy as np
import pandas as pd
import requests
from io import BytesIO
from PIL import Image
import os
from colorthief import ColorThief
import pytesseract
from multiprocessing import Pool
from cnsenti import Sentiment
import pynlpir
from collections import defaultdict
import warnings

warnings.filterwarnings('ignore')

#设置OCR路径
pytesseract.pytesseract.tesseract_cmd = r'D:Program files\Tesseract-OCR\tesseract.exe'

# ------------------图像处理-初始化配置 ---------------------
# 人脸检测模型初始化
face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')

# 图像情感模型系数（基于IAPS数据集校准）
VALENCE_WEIGHTS = {
    'warm_ratio': 0.35,
    'brightness': 0.15,
    'symmetry': 0.20,
    'colorfulness': 0.30
}

AROUSAL_WEIGHTS = {
    'contrast': 0.40,
    'edge_density': 0.35,
    'saturation_std': 0.25
}

# 暖色调定义（HSV色相范围）
WARM_HUE_RANGE = (0, 60)  # 红色到黄色

# ------------------处理图像 ---------------------
def get_image(url):
    """从URL获取图像并预处理"""
    try:
        response = requests.get(url, timeout=10)
        img = Image.open(BytesIO(response.content))
        return cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR)
    except Exception as e:
        print(f"Error loading {url}: {str(e)}")
        return None


def analyze_image(img):
    """分析图像特征"""
    if img is None:
        return {}

    # 转换为HSV颜色空间
    hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
    h, s, v = cv2.split(hsv)

    # 计算基础特征
    features = {
        'brightness': np.mean(v),
        'contrast': np.max(v) - np.min(v),
        'saturation_std': np.std(s),
        'colorfulness': np.std(h) + np.std(s) + np.std(v)
    }

    # 暖色比例计算
    hue_mask = cv2.inRange(h, WARM_HUE_RANGE[0], WARM_HUE_RANGE[1])
    features['warm_ratio'] = np.count_nonzero(hue_mask) / (img.shape[0] * img.shape[1])

    # 对称性计算
    mid = img.shape[1] // 2
    left = img[:, :mid]
    right = cv2.flip(img[:, mid:], 1)
    features['symmetry'] = cv2.matchTemplate(left, right, cv2.TM_CCOEFF_NORMED)[0][0]

    # 边缘密度
    edges = cv2.Canny(cv2.cvtColor(img, cv2.COLOR_BGR2GRAY), 100, 200)
    features['edge_density'] = np.mean(edges)

    return features


def calculate_affect(features):
    """计算情感效价和唤醒度"""
    poslm = sum(features[k] * VALENCE_WEIGHTS[k] for k in VALENCE_WEIGHTS)
    actlm = sum(features[k] * AROUSAL_WEIGHTS[k] for k in AROUSAL_WEIGHTS)

    # Sigmoid归一化
    return {
        'Poslm': 2 / (1 + np.exp(-poslm)) - 1,
        'Actlm': 2 / (1 + np.exp(-actlm)) - 1
    }


def detect_human(img):
    """检测人像"""
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    faces = face_cascade.detectMultiScale(gray, 1.05, 3)
    return len(faces) > 0


def process_url(url):
    try:
        img = get_image(url)
        if img is None:
            return None

        features = analyze_image(img)
        affect = calculate_affect(features)

        # 主色分析
        color_thief = ColorThief(BytesIO(requests.get(url).content))
        dominant_color = color_thief.get_color(quality=1)
        hsv_color = cv2.cvtColor(np.uint8([[dominant_color]]), cv2.COLOR_RGB2HSV)[0][0]
        warm = 1 if WARM_HUE_RANGE[0] <= hsv_color[0] <= WARM_HUE_RANGE[1] else 0

        # 人像检测
        has_human = detect_human(img)

        return {
            'url': url,
            **affect,
            'Warm': warm,
            'Portrait': int(has_human)
        }
    except Exception as e:
        print(f"Error processing {url}: {str(e)}")
        return None


# 批量处理
def batch_process(urls, workers=4):
    with Pool(workers) as pool:
        results = [res for res in pool.imap(process_url, urls) if res is not None]
    return pd.DataFrame(results)


# 使用示例
if __name__ == "__main__":
    # 读取URL列表
    input_csv = "data_all.csv"
    #输出路径
    os.makedirs('./result', exist_ok=True)
    output_csv = "result/analysis_results.csv"

    ##完整运行
    # df = pd.read_csv(input_csv)
    # urls = df['视频封面'].tolist()
    #
    # # 执行分析
    # result_df = batch_process(urls)
    #
    # # 合并原始数据
    # final_df = df.merge(result_df, left_on='视频封面')
    # final_df.drop('url', axis=1).to_csv(output_csv, index=False)

    # 示例URL列表
    #小批量实验
    urls = [
        'http://i0.hdslb.com/bfs/archive/393a8e961b704d43256fe7e6c89fee04df966e17.jpg',
        'http://i0.hdslb.com/bfs/archive/072e16a1237040941f15b1ed67a8d1ebe6f2e041.jpg',
        'http://i2.hdslb.com/bfs/archive/1c56b5bec767c604175983cc5926f5832baa9bb8.jpg',
        'http://i0.hdslb.com/bfs/archive/66384e53a15345a539ccbb2989442f1d960b9235.jpg',
        'http://i2.hdslb.com/bfs/archive/836b762456f0b4d65dd2c40fc4cd120107e46b88.jpg',
    ]
    result_df = batch_process(urls)
    result_df.to_csv("result/analysis_results.csv", index=False)
    print(f"成功处理 {len(result_df)}/{len(urls)} 张图片")


    print("分析完成！结果已保存至", output_csv)