Анализ видеоконтента

VK Video Analytics — Анализ видеоконтента

Автор: Светлана Селиверстова

Группа: АДЭУ-221

Дисциплина: Работа с API социальных сетей и облачных сервисов

Цель работы

Разработать инструмент для анализа видеоконтента из открытых источников (VK) с целью выявления актуальных тем, изучения принципов работы API и создания интерактивного веб-интерфейса для визуализации метрик.

Задачи

1. Изучить документацию VK API (методы video.get, users.get, groups.getById, utils.resolveScreenName).
2. Получить токен доступа для работы с данными.
3. Написать скрипт на Python для сбора до 500 видео из сообщества.
4. Реализовать фильтрацию «лишних» слов (местоимения, глаголы, вводные конструкции) для выделения смысловых тем.
5. Построить графики: топ-10 видео, анализ длительности, временная активность, облако ключевых слов, вовлечённость.
6. Создать веб-интерфейс на Flask для отображения аналитики в браузере (localhost).
7. Оформить результаты в виде отчёта.

Технологии

• Языки: Python, JavaScript, HTML/CSS
• Библиотеки Python: requests, pandas, matplotlib, seaborn, wordcloud, flask
• API: VK API (видео, пользователи, группы, разрешение имён)
• Визуализация: Matplotlib, Seaborn, WordCloud
• Веб-фреймворк: Flask

Структура проекта

vk_video_analytics/
├── config.py                 # настройки: токен, ссылка, лимиты
├── vk_api_client.py          # работа с VK API (пагинация, resolveScreenName)
├── analyzer.py               # аналитика: DataFrame, стоп-слова, тренды
├── visualizer.py             # построение графиков (matplotlib, wordcloud)
├── app.py                    # веб-сервер Flask
├── templates/
│   └── index.html            # HTML-шаблон с графиками и трендами
├── results/                  # папка с результатами (создаётся автоматически)
│   ├── videos_data.csv       # все собранные данные
│   ├── *.png                 # графики
│   └── *.json                # инсайты и тренды
└── requirements.txt          # зависимости

Ход работы над проектом

Этап 1. Написание Python-скрипта для сбора данных

Установила необходимые библиотеки:

pip install requests pandas matplotlib seaborn wordcloud flask

Написала клиент для VK API с поддержкой:

• Преобразования ссылки в ID через utils.resolveScreenName
• Пагинации для получения до 500 видео
• Сбора информации об авторах (пользователи и группы)

Результат: собран DataFrame с полями: title, views, duration, date, engagement и др.

Этап 2. Реализация фильтрации «лишних» слов

Создала список стоп-слов, включающий:

• Местоимения (твой, мой, его, их, я, ты, мы)
• Глаголы (быть, сказать, делать, хотеть, любить, идти, работать)
• Вводные конструкции (кстати, наверное, конечно, по-моему)
• Частицы, предлоги, союзы (в, на, и, а, но, или)
• Общие слова (самый, хороший, новый, легко, сейчас)

При построении облака слов и поиске трендовых тем эти слова исключаются, остаются только смысловые ключевые слова.

Этап 3. Создание графиков и визуализации

Реализовала модуль визуализации, который строит:

• Топ-10 видео по просмотрам (горизонтальная столбчатая диаграмма)
• Анализ длительности — средние просмотры и распределение по категориям
• Временной анализ — средние просмотры по часам и дням недели
• Облако ключевых слов (только значимые слова после фильтрации)
• Анализ вовлечённости — зависимость вовлечённости от просмотров и по длительности

Этап 4. Создание веб-интерфейса на Flask

Разработала веб-приложение на Flask с маршрутами:

• / — главная страница с аналитикой
• /api/insights — JSON с ключевыми инсайтами
• /api/trending — JSON с трендовыми темами
• /api/top_videos — JSON с топ-20 видео
• /api/chart/<name> — отдача PNG-графиков
• /refresh — принудительное обновление данных

HTML-шаблон адаптивный, отображает все графики и тренды в удобной сетке.

Этап 5. Тестирование и отладка

Сервис доступен по адресу http://localhost:5000/ Провела тестирование на сообществе https://vk.com/vkvideo. Собрано до 500 видео, построены графики. Проверила фильтрацию стоп-слов — в облаке слов и трендах остались только смысловые темы (например, «кино», «сериал», «премьера», «трейлер»).

Выводы

В ходе работы над проектом я:

• Научилась работать с VK API — получать токен, делать запросы с пагинацией, обрабатывать ответы, использовать resolveScreenName для преобразования ссылок.
• Освоила очистку текстовых данных: фильтрацию стоп-слов (местоимения, глаголы, вводные конструкции) для выделения смысловых тем.
• Создала комплексную аналитику видеоконтента: топ видео, анализ длительности, временную активность, вовлечённость, облако ключевых слов.
• Разработала веб-приложение на Flask с интерактивным интерфейсом и графиками.
• Поняла, как устроены современные сервисы аналитики контента, собирающие данные через открытые API.

Интерфейс проекта

Код проекта

config.py

# config.py - настройки проекта
VK_TOKEN = "your_token_here"
COMMUNITY_INPUT = "https://vk.com/vkvideo"
VIDEOS_LIMIT = 500
RESULTS_DIR = "results"

app.py

from flask import Flask, render_template, jsonify, send_file
import os
import json
from datetime import datetime
from config import VK_TOKEN, COMMUNITY_INPUT, VIDEOS_LIMIT, RESULTS_DIR
from vk_api_client import VKAPIClient
from analyzer import VideoAnalyzer
from visualizer import VideoVisualizer

app = Flask(__name__)

analysis_results = None
analyzer_instance = None
community_id = None

def get_community_id():
    global community_id
    if community_id is not None:
        return community_id
    client = VKAPIClient()
    inp = COMMUNITY_INPUT.strip()
    if inp.lstrip('-').isdigit():
        community_id = int(inp)
        return community_id
    resolved = client.resolve_screen_name(inp)
    if resolved and resolved.get('type') == 'group':
        community_id = -resolved['object_id']
        return community_id
    return None

def run_analysis():
    global analysis_results, analyzer_instance
    print("📊 Запуск анализа...")
    client = VKAPIClient()
    cid = get_community_id()
    if not cid:
        print("❌ Сообщество не найдено")
        return None

    print(f"📹 Получение видео из сообщества {cid} (макс. {VIDEOS_LIMIT})...")
    videos = client.get_community_videos(cid, VIDEOS_LIMIT)
    if not videos:
        print("❌ Видео не получены")
        return None

    author_ids = list(set(abs(v['owner_id']) for v in videos))
    user_ids = [uid for uid in author_ids if uid > 0]
    group_ids = [uid for uid in author_ids if uid < 0]

    authors_data = {}
    if user_ids:
        authors_data.update(client.get_user_info(user_ids))
    if group_ids:
        authors_data.update(client.get_group_info(group_ids))

    analyzer = VideoAnalyzer(videos, authors_data)
    analyzer_instance = analyzer

    os.makedirs(RESULTS_DIR, exist_ok=True)
    analyzer.df.to_csv(f"{RESULTS_DIR}/videos_data.csv", index=False, encoding='utf-8-sig')

    vis = VideoVisualizer(analyzer, RESULTS_DIR)
    vis.create_full_report()

    insights = analyzer.get_insights()
    trending = analyzer.get_trending_topics(20)

    with open(f"{RESULTS_DIR}/insights.json", 'w', encoding='utf-8') as f:
        json.dump(insights, f, ensure_ascii=False, indent=2, default=str)
    with open(f"{RESULTS_DIR}/trending_topics.json", 'w', encoding='utf-8') as f:
        json.dump(trending, f, ensure_ascii=False, indent=2)

    analysis_results = {
        'insights': insights,
        'trending': trending,
        'videos_count': len(videos),
        'analyzed_at': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        'community_input': COMMUNITY_INPUT
    }
    print("✅ Анализ завершён")
    return analysis_results

@app.route('/')
def index():
    global analysis_results
    if analysis_results is None:
        analysis_results = run_analysis()
    if analysis_results is None:
        return "Ошибка: проверьте токен и ссылку на сообщество", 500
    return render_template('index.html', results=analysis_results)

@app.route('/api/insights')
def api_insights():
    global analysis_results
    if analysis_results is None:
        analysis_results = run_analysis()
    return jsonify(analysis_results.get('insights', {}))

@app.route('/api/trending')
def api_trending():
    global analysis_results
    if analysis_results is None:
        analysis_results = run_analysis()
    return jsonify(analysis_results.get('trending', []))

@app.route('/api/top_videos')
def api_top_videos():
    global analyzer_instance
    if analyzer_instance is None:
        run_analysis()
    top = analyzer_instance.get_top_videos(20)
    return jsonify(top.to_dict('records'))

@app.route('/api/chart/<name>')
def get_chart(name):
    chart_map = {
        'views': 'views_distribution.png',
        'duration': 'duration_analysis.png',
        'temporal': 'temporal_analysis.png',
        'trending': 'trending_topics.png',
        'engagement': 'engagement_analysis.png'
    }
    if name in chart_map:
        path = os.path.join(RESULTS_DIR, chart_map[name])
        if os.path.exists(path):
            return send_file(path, mimetype='image/png')
    return "Not found", 404

@app.route('/refresh')
def refresh():
    global analysis_results, analyzer_instance
    analysis_results = None
    analyzer_instance = None
    run_analysis()
    return "OK", 200

if __name__ == '__main__':
    print("="*60)
    print("🎬 VK Video Analytics")
    print(f"Ссылка на сообщество: {COMMUNITY_INPUT}")
    print("Сервер: http://localhost:5000")
    print("="*60)
    app.run(debug=True, host='0.0.0.0', port=5000)

analyzer.py

import pandas as pd
import numpy as np
from datetime import datetime
from collections import Counter
import re
from typing import List, Dict

class VideoAnalyzer:
    # Полный обновлённый список стоп-слов
    STOP_WORDS = {
        # ========== 1. Притяжательные и личные местоимения ==========
        'твой', 'твоя', 'твоё', 'твои', 'мой', 'моя', 'моё', 'мои', 'наш', 'наша', 'наше', 'наши',
        'ваш', 'ваша', 'ваше', 'ваши', 'его', 'её', 'их', 'свой', 'своя', 'своё', 'свои', 'я', 'ты',
        'мы', 'вы', 'он', 'она', 'оно', 'они',
        
        # ========== 2. Глаголы (основные формы) ==========
        # Бытовые и вспомогательные
        'быть', 'мочь', 'сказать', 'говорить', 'знать', 'делать', 'хотеть', 'любить', 'становиться',
        'являться', 'принимать', 'спрашивать', 'думать', 'полагать', 'надеяться', 'понимать', 'видеть',
        'слышать', 'помнить', 'считать', 'выглядеть', 'происходить', 'случаться', 'казаться', 'получаться',
        'получиться', 'идти', 'приходить', 'уходить', 'ехать', 'приезжать', 'работать', 'жить', 'есть',
        'иметь', 'давать', 'брать', 'получать', 'начинать', 'заканчивать', 'продолжать', 'пробовать',
        'пытаться', 'стараться', 'успевать', 'успеть', 'ждать', 'смотреть', 'слушать',
        # Дополнительные частые глаголы
        'рассказывать', 'показывать', 'объяснять', 'учить', 'учиться', 'играть', 'писать', 'читать',
        'пить', 'спать', 'вставать', 'ложиться', 'сидеть', 'стоять', 'лежать', 'бегать', 'ходить',
        'плавать', 'лететь', 'падать', 'садиться',
        # Глаголы в повелительном наклонении
        'послушай', 'послушайте', 'посмотри', 'посмотрите', 'представь', 'вообрази', 'поверь', 'пойми',
        'скажи', 'сделай', 'попробуй', 'давай', 'давайте',
        
        # ========== 3. Вводные слова и конструкции ==========
        # Уверенность / неуверенность
        'конечно', 'разумеется', 'несомненно', 'безусловно', 'бесспорно', 'действительно', 'наверное',
        'вероятно', 'возможно', 'может', 'должно', 'кажется', 'пожалуй', 'видимо', 'очевидно', 'правда',
        'само', 'собой', 'без', 'сомнения', 'в самом деле',
        # Связь мыслей
        'кстати', 'кстати сказать', 'к слову', 'итак', 'следовательно', 'значит', 'таким образом',
        'во-первых', 'во-вторых', 'в-третьих', 'наконец', 'далее', 'впрочем', 'напротив', 'наоборот',
        'в общем', 'в частности', 'кроме того', 'между прочим', 'главное', 'стало быть', 'например',
        'к примеру',
        # Эмоции и оценка
        'к счастью', 'к сожалению', 'к удивлению', 'к несчастью', 'к радости', 'на беду', 'к ужасу',
        'к стыду', 'чего доброго', 'не дай бог', 'слава богу', 'на радость', 'к моему стыду',
        # Источник сообщения
        'по-моему', 'по-твоему', 'по-нашему', 'по-вашему', 'говорят', 'по словам', 'по мнению',
        'по слухам', 'помнится', 'дескать', 'мол', 'как известно', 'слышно', 'сообщают',
        # Оформление мыслей
        'словом', 'одним словом', 'иными словами', 'так сказать', 'собственно говоря', 'вообще говоря',
        'вернее сказать', 'мягко выражаясь', 'видите ли', 'знаешь ли', 'понимаешь ли', 'поверьте',
        'вообразите', 'представьте', 'согласитесь', 'пожалуйста',
        # Другие вводные
        'в конце концов', 'в общем-то', 'на самом деле', 'в то же время', 'между тем', 'фактически',
        'буквально', 'обычно', 'как правило', 'по обыкновению', 'бывало', 'однажды',
        
        # ========== 4. Частицы, предлоги, союзы, прочая «вода» ==========
        # Частицы и усилители
        'как', 'что', 'это', 'вот', 'же', 'ли', 'не', 'ни', 'бы', 'ведь', 'уже', 'еще', 'только',
        'просто', 'прямо', 'именно', 'вовсе', 'совсем', 'очень', 'довольно', 'почти', 'примерно',
        'точно', 'реально', 'вообще', 'здесь', 'тут', 'там', 'этот', 'эта', 'эти', 'тот', 'та', 'то',
        'те', 'такой', 'такая', 'такое', 'такие', 'один', 'одна', 'одно', 'одни', 'какой', 'какая',
        'какое', 'какие', 'который', 'которая', 'которое', 'которые', 'весь', 'вся', 'всё', 'все',
        'любой', 'любая', 'любое', 'любые', 'некоторый', 'некоторая', 'некоторое', 'некоторые',
        'самый', 'самая', 'самое', 'самые', 'лучший', 'лучшая', 'лучшее', 'лучшие', 'хороший',
        'хорошая', 'хорошее', 'хорошие', 'новый', 'новая', 'новое', 'новые', 'легко', 'быстро',
        'бесплатно', 'сейчас', 'сегодня', 'завтра', 'всегда', 'никогда', 'можно', 'нужно', 'должен',
        'должна', 'должны',
        # Предлоги и союзы
        'в', 'во', 'на', 'по', 'из', 'от', 'до', 'для', 'за', 'под', 'над', 'перед', 'при', 'о', 'об',
        'про', 'через', 'у', 'с', 'со', 'без', 'и', 'а', 'но', 'или', 'если', 'когда', 'то', 'тогда',
        'чтобы', 'потому', 'что', 'так', 'как', 'потому что', 'так как'
    }

    def __init__(self, videos_data: List[Dict], authors_data: Dict):
        self.videos_data = videos_data
        self.authors_data = authors_data
        self.df = self._create_dataframe()

    def _create_dataframe(self) -> pd.DataFrame:
        rows = []
        for video in self.videos_data:
            title = video.get('title', '')
            hashtags = [w.lower() for w in title.split() if w.startswith('#')]
            owner_id = video.get('owner_id', 0)
            author_id = abs(owner_id)
            author_info = self.authors_data.get(author_id, {})
            likes = video.get('likes', {})
            likes_count = likes.get('count', 0) if isinstance(likes, dict) else int(likes) if likes else 0
            reposts = video.get('reposts', {})
            reposts_count = reposts.get('count', 0) if isinstance(reposts, dict) else int(reposts) if reposts else 0

            row = {
                'video_id': video.get('id'),
                'title': title,
                'title_clean': re.sub(r'[#\W]+', ' ', title).strip().lower(),
                'hashtags': hashtags,
                'duration': video.get('duration', 0),
                'views': video.get('views', 0),
                'comments': video.get('comments', 0),
                'likes': likes_count,
                'reposts': reposts_count,
                'date': datetime.fromtimestamp(video.get('date', 0)),
                'owner_id': owner_id,
                'author_id': author_id,
                'author_name': author_info.get('name', author_info.get('first_name', str(author_id))),
                'author_followers': author_info.get('followers_count', author_info.get('members_count', 0)),
            }
            row['engagement'] = row['likes'] + row['comments'] + row['reposts']
            row['engagement_rate'] = row['engagement'] / max(row['views'], 1)
            rows.append(row)

        df = pd.DataFrame(rows)
        df['duration_category'] = pd.cut(
            df['duration'],
            bins=[0, 60, 300, 1800, float('inf')],
            labels=['short (0-1 min)', 'medium (1-5 min)', 'long (5-30 min)', 'very long (>30 min)']
        )
        return df

    def analyze_by_duration(self) -> pd.DataFrame:
        return self.df.groupby('duration_category').agg(
            avg_views=('views', 'mean'),
            videos_count=('video_id', 'count')
        )

    def get_trending_topics(self, limit=10) -> List[Dict]:
        # Собираем слова из названий (очищенных от хэштегов и пунктуации)
        words = []
        for title in self.df['title_clean']:
            for w in title.split():
                # удаляем стоп-слова и короткие слова
                if len(w) > 3 and w not in self.STOP_WORDS:
                    words.append(w)
        word_count = Counter(words)

        trending = []
        for word, cnt in word_count.most_common(50):
            videos = self.df[self.df['title_clean'].str.contains(word, na=False)]
            if len(videos) > 2:
                avg_views = videos['views'].mean()
                # трендовый балл = 30% частоты + 70% нормализованных просмотров
                score = (cnt * 0.3) + (avg_views / self.df['views'].max() * 0.7)
                trending.append({
                    'topic': word,
                    'type': 'word',
                    'frequency': cnt,
                    'avg_views': int(avg_views),
                    'trending_score': round(score, 3),
                    'videos_with_topic': len(videos)
                })

        trending.sort(key=lambda x: x['trending_score'], reverse=True)
        return trending[:limit]

    def get_temporal_analysis(self) -> Dict:
        df = self.df.copy()
        df['hour'] = df['date'].dt.hour
        df['weekday'] = df['date'].dt.day_name()
        hourly = df.groupby('hour')['views'].mean().round(0)
        weekday_order = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
        weekday = df.groupby('weekday')['views'].mean().reindex(weekday_order).round(0)
        return {'hourly': hourly, 'weekday': weekday}

    def get_top_videos(self, limit=10) -> pd.DataFrame:
        return self.df.nlargest(limit, 'views')[['title', 'views', 'engagement', 'duration_category', 'date']]

    def get_insights(self) -> Dict:
        insights = {
            'total_videos': len(self.df),
            'total_views': int(self.df['views'].sum()),
            'avg_views': int(self.df['views'].mean()),
            'max_views': int(self.df['views'].max()),
            'best_duration': self.df.groupby('duration_category')['views'].mean().idxmax(),
            'best_hour': int(self.df.groupby(self.df['date'].dt.hour)['views'].mean().idxmax()),
            'best_weekday': self.df.groupby(self.df['date'].dt.day_name())['views'].mean().idxmax(),
        }
        return insights