philosophy_moderator/ ├── config.py # настройки секций и параметров ├── semiotic_analyzer.py # ядро семиотического анализа ├── database.py # работа с БД (история авторов) ├── moderator_bot.py # основной скрипт-модератор ├── web_interface.py # веб-интерфейс для модераторов └── requirements.txt # зависимости

# config.py # Конфигурация семиотического модератора # Секции форума SECTIONS = { "analytical": { "name": "Аналитическая философия", "allowed_signs": [[0,0,0], [1,0,0], [0,0,1], [1,0,1]], "description": "Для фактов, определений, доказательств" }, "expressive": { "name": "Экспрессивная философия", "allowed_signs": [[0,-1,0], [0,-1,-1], [0,-1,1]], "description": "Для восклицаний, образов, свободного потока" }, "schematic": { "name": "Схематическая философия", "allowed_signs": [[-1,0.5,-1], [-1,0.5,0], [-1,0.5,1]], "description": "Для гипотетической полемики, выявления противоречий" }, "hermetic": { "name": "Герметический дискурс", "allowed_signs": [[0,0.5,-1], [0,0.5,0], [0,0.5,1]], "description": "Для герметичных, метафизических текстов" }, "general": { "name": "Общее пространство", "allowed_signs": "all", "description": "Для всех типов речи (по умолчанию)" } } # Параметры анализа COORDINATE_RANGES = { "sense": [-1, -0.5, 0, 0.5, 1], # смысл/польза "form": [-1, -0.5, 0, 0.5, 1], # форма высказывания "cause": [-1, -0.5, 0, 0.5, 1] # причина высказывания } # Порог совместимости для публикации в другой секции COMPATIBILITY_THRESHOLD = 0.7 # Временное окно для установления тренда (количество сообщений) TREND_WINDOW_SIZE = 5 # Режим работы: "test", "warning", "active" MODE = "test" # Настройки базы данных DATABASE = { "type": "sqlite", "path": "moderator.db" }

# semiotic_analyzer.py # Ядро семиотического анализа текста import re import math from typing import List, Tuple, Dict, Optional class SemioticAnalyzer: """ Семиотический анализатор текста. Вычисляет три координаты: смысл, форма, причина. """ def __init__(self): # Маркеры для координаты 1 (смысл/польза) self.self_markers = [ r'я (считаю|думаю|полагаю|уверен)', r'мо[ейя]', r'меня', r'мне', r'по-моему', r'на мой взгляд' ] self.attack_markers = [ r'вы (не правы|ошибаетесь|не понимаете)', r'ваш[а-я]? (ошибка|заблуждение)', r'ты (не прав|ошибаешься)' ] self.fact_markers = [ r'это (факт|истина|так)', r'безусловно', r'несомненно', r'очевидно, что' ] # Маркеры для координаты 2 (форма) self.literal_markers = [ r'буквально', r'именно', r'в точности', r'фактически' ] self.symbolic_markers = [ r'символизирует', r'означает', r'метафора', r'образ', r'как символ' ] self.substitution_markers = [ r'это и есть', r'то же, что', r'сводится к', r'по сути' ] # Маркеры для координаты 3 (причина) self.hypothesis_markers = [ r'возможно', r'предположим', r'если допустить', r'может быть', r'вероятно' ] self.question_markers = [ r'\?', r'разве', r'неужели', r'что, если' ] self.proof_markers = [ r'следовательно', r'таким образом', r'из этого следует', r'доказывает', r'поскольку.*то' ] self.narrative_markers = [ r'расскажу', r'опишу', r'представьте', r'однажды' ] def analyze(self, text: str) -> Dict[str, float]: """ Анализирует текст и возвращает три координаты. """ text = text.lower() sense = self._calculate_sense(text) form = self._calculate_form(text) cause = self._calculate_cause(text) return { "sense": sense, "form": form, "cause": cause, "sign": [sense, form, cause] } def _calculate_sense(self, text: str) -> float: """Вычисляет координату смысла/пользы.""" score = 0.0 # Проверка маркеров "себе" (польза себе) for pattern in self.self_markers: if re.search(pattern, text): score -= 0.3 # Проверка маркеров атаки на оппонента for pattern in self.attack_markers: if re.search(pattern, text): score -= 0.4 # Проверка маркеров факта for pattern in self.fact_markers: if re.search(pattern, text): score += 0.3 # Анализ императивов (приказов, требований) imperatives = re.findall(r'[!！]', text) if len(imperatives) > 2: score -= 0.2 # Анализ местоимений if re.search(r'\b(ты|вы)\b', text) and not re.search(r'\bмы\b', text): score -= 0.2 # Нормализация к диапазону [-1, 1] score = max(-1.0, min(1.0, score)) # Округление до ближайшего допустимого значения return self._round_to_nearest(score, [-1, -0.5, 0, 0.5, 1]) def _calculate_form(self, text: str) -> float: """Вычисляет координату формы высказывания.""" score = 0.0 # Буквальные маркеры for pattern in self.literal_markers: if re.search(pattern, text): score -= 0.3 # Символические маркеры for pattern in self.symbolic_markers: if re.search(pattern, text): score += 0.3 # Маркеры подмены for pattern in self.substitution_markers: if re.search(pattern, text): if score < 0: score = -0.5 # подмена символическим буквального else: score = 0.5 # подмена буквальным абстрактного # Анализ метафор (по наличию сравнительных оборотов) metaphors = re.findall(r'как\s+\w+', text) if len(metaphors) > 1: score = 0.0 # Анализ восклицаний (экспрессия) exclamations = re.findall(r'[!！]', text) if len(exclamations) > 0: score -= 0.2 # Нормализация score = max(-1.0, min(1.0, score)) # Округление return self._round_to_nearest(score, [-1, -0.5, 0, 0.5, 1]) def _calculate_cause(self, text: str) -> float: """Вычисляет координату причины высказывания.""" score = 0.0 # Гипотезы for pattern in self.hypothesis_markers: if re.search(pattern, text): score -= 0.3 # Вопросы if re.search(self.question_markers[0], text): score -= 0.2 for pattern in self.question_markers[1:]: if re.search(pattern, text): score -= 0.3 # Доказательства for pattern in self.proof_markers: if re.search(pattern, text): score += 0.4 # Нарратив for pattern in self.narrative_markers: if re.search(pattern, text): score = 0.0 # Нормализация score = max(-1.0, min(1.0, score)) # Округление return self._round_to_nearest(score, [-1, -0.5, 0, 0.5, 1]) def _round_to_nearest(self, value: float, allowed: List[float]) -> float: """Округляет значение до ближайшего допустимого.""" if not allowed: return value return min(allowed, key=lambda x: abs(x - value)) def get_sign_name(self, sign: List[float]) -> str: """Возвращает название семиотического знака.""" sign_map = { (0,0,0): "Факт", (0,-1,0): "Неопределённое восклицание", (-1,0.5,-1): "Схема", (0,0.5,-1): "Герметическая схема", (1,0,0): "Всеобщее утверждение", (0,0,1): "Доказательство", (-1,0,0): "Частное утверждение" } key = tuple(sign) return sign_map.get(key, "Неопределённый знак") def get_recommended_section(self, sign: List[float]) -> str: """Рекомендует секцию на основе семиотического знака.""" sign_tuple = tuple(sign) if sign_tuple in [(0,0,0), (1,0,0), (0,0,1), (1,0,1)]: return "analytical" elif sign_tuple in [(0,-1,0), (0,-1,-1), (0,-1,1)]: return "expressive" elif sign_tuple in [(-1,0.5,-1), (-1,0.5,0), (-1,0.5,1)]: return "schematic" elif sign_tuple in [(0,0.5,-1), (0,0.5,0), (0,0.5,1)]: return "hermetic" else: return "general"

# database.py # Работа с базой данных пользователей и их семиотических профилей import sqlite3 import json from typing import List, Dict, Optional from datetime import datetime, timedelta class Database: def __init__(self, db_path: str = "moderator.db"): self.db_path = db_path self._init_tables() def _init_tables(self): """Инициализация таблиц базы данных.""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() # Таблица сообщений cursor.execute(""" CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, message_text TEXT NOT NULL, timestamp TIMESTAMP NOT NULL, sense REAL, form REAL, cause REAL, recommended_section TEXT, published_section TEXT, status TEXT DEFAULT 'pending' ) """) # Таблица семиотических профилей пользователей cursor.execute(""" CREATE TABLE IF NOT EXISTS user_profiles ( user_id INTEGER PRIMARY KEY, sign_history TEXT NOT NULL, dominant_sign TEXT, trend_stable BOOLEAN DEFAULT TRUE, current_section TEXT, last_updated TIMESTAMP ) """) # Таблица предупреждений и нарушений cursor.execute(""" CREATE TABLE IF NOT EXISTS violations ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, message_id INTEGER, violation_type TEXT, timestamp TIMESTAMP ) """) # Таблица для статистики cursor.execute(""" CREATE TABLE IF NOT EXISTS statistics ( date DATE PRIMARY KEY, total_messages INTEGER, conflicts INTEGER, section_distribution TEXT ) """) conn.commit() def save_message_analysis(self, user_id: int, message_text: str, coordinates: Dict, section: str, status: str = "pending"): """Сохраняет результат анализа сообщения.""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO messages (user_id, message_text, timestamp, sense, form, cause, recommended_section, published_section, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( user_id, message_text, datetime.now(), coordinates["sense"], coordinates["form"], coordinates["cause"], section, section, status )) conn.commit() return cursor.lastrowid def update_user_profile(self, user_id: int, sign: List[float]): """Обновляет семиотический профиль пользователя.""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() # Получаем текущую историю cursor.execute( "SELECT sign_history FROM user_profiles WHERE user_id = ?", (user_id,) ) row = cursor.fetchone() if row: history = json.loads(row[0]) history.append({ "timestamp": datetime.now().isoformat(), "sign": sign }) # Оставляем только последние N записей if len(history) > 20: history = history[-20:] dominant = self._calculate_dominant_sign(history) trend_stable = self._check_trend_stability(history) cursor.execute(""" UPDATE user_profiles SET sign_history = ?, dominant_sign = ?, trend_stable = ?, last_updated = ? WHERE user_id = ? """, (json.dumps(history), json.dumps(dominant), trend_stable, datetime.now(), user_id)) else: # Новый пользователь history = [{"timestamp": datetime.now().isoformat(), "sign": sign}] cursor.execute(""" INSERT INTO user_profiles (user_id, sign_history, dominant_sign, trend_stable, last_updated) VALUES (?, ?, ?, ?, ?) """, (user_id, json.dumps(history), json.dumps(sign), True, datetime.now())) conn.commit() def _calculate_dominant_sign(self, history: List) -> List[float]: """Вычисляет доминирующий семиотический знак пользователя.""" if not history: return [0, 0, 0] sums = [0.0, 0.0, 0.0] for entry in history[-10:]: # последние 10 сообщений sign = entry["sign"] sums[0] += sign[0] sums[1] += sign[1] sums[2] += sign[2] count = len(history[-10:]) return [s / count for s in sums] def _check_trend_stability(self, history: List) -> bool: """Проверяет стабильность тренда.""" if len(history) < 3: return True recent = history[-3:] signs = [entry["sign"] for entry in recent] # Проверяем отклонение for i in range(3): if abs(signs[0][i] - signs[-1][i]) > 0.5: return False return True def get_user_section(self, user_id: int) -> Optional[str]: """Возвращает текущую секцию пользователя.""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute( "SELECT current_section FROM user_profiles WHERE user_id = ?", (user_id,) ) row = cursor.fetchone() return row[0] if row else None def log_violation(self, user_id: int, message_id: int, violation_type: str): """Логирует нарушение (попытка публикации в несовместимой секции).""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO violations (user_id, message_id, violation_type, timestamp) VALUES (?, ?, ?, ?) """, (user_id, message_id, violation_type, datetime.now())) conn.commit()

# moderator_bot.py # Основной скрипт-модератор import sys from datetime import datetime from typing import Dict, Optional from config import MODE, COMPATIBILITY_THRESHOLD, TREND_WINDOW_SIZE from semiotic_analyzer import SemioticAnalyzer from database import Database class ModeratorBot: def __init__(self): self.analyzer = SemioticAnalyzer() self.db = Database() self.mode = MODE def process_message(self, user_id: int, message_text: str, target_section: str = "general") -> Dict: """ Обрабатывает сообщение пользователя. Возвращает решение: разрешить публикацию или перенаправить. """ # 1. Анализируем сообщение analysis = self.analyzer.analyze(message_text) sign = analysis["sign"] recommended_section = self.analyzer.get_recommended_section(sign) # 2. Сохраняем анализ в БД message_id = self.db.save_message_analysis( user_id, message_text, analysis, recommended_section, "analyzed" ) # 3. Обновляем профиль пользователя self.db.update_user_profile(user_id, sign) # 4. Проверяем совместимость с целевой секцией is_compatible = self._check_compatibility( sign, recommended_section, target_section ) # 5. Принимаем решение if self.mode == "test": # Тестовый режим: только логируем, не блокируем return { "allowed": True, "mode": "test", "sign": sign, "sign_name": self.analyzer.get_sign_name(sign), "recommended_section": recommended_section, "message": f"[ТЕСТ] Ваше сообщение имеет знак {sign} " f"({self.analyzer.get_sign_name(sign)}). " f"Рекомендуемая секция: {recommended_section}" } elif self.mode == "warning": # Режим предупреждений if not is_compatible: self.db.log_violation(user_id, message_id, "section_mismatch") return { "allowed": True, # всё равно разрешаем, но с предупреждением "mode": "warning", "sign": sign, "recommended_section": recommended_section, "message": f"⚠️ ВАЖНО: Ваше сообщение ({self.analyzer.get_sign_name(sign)}) " f"лучше соответствует секции «{recommended_section}». " f"Публикация в «{target_section}» может вызвать непонимание." } return { "allowed": True, "mode": "warning", "sign": sign, "message": "Сообщение совместимо с выбранной секцией." } else: # active mode # Активный режим: блокируем несовместимые сообщения if not is_compatible: self.db.log_violation(user_id, message_id, "section_mismatch") # Проверяем историю нарушений if self._has_repeated_violations(user_id): return { "allowed": False, "mode": "blocked", "sign": sign, "recommended_section": recommended_section, "message": f"❌ СООБЩЕНИЕ ЗАБЛОКИРОВАНО.\n" f"Ваш тип речи ({self.analyzer.get_sign_name(sign)}) " f"не совместим с секцией «{target_section}».\n" f"Рекомендуется публиковать в секции «{recommended_section}».\n" f"Повторные попытки приведут к временной изоляции." } else: return { "allowed": False, "mode": "redirect", "sign": sign, "recommended_section": recommended_section, "message": f"⚠️ Сообщение не совместимо с секцией «{target_section}».\n" f"Ваш тип речи: {self.analyzer.get_sign_name(sign)}.\n" f"Пожалуйста, опубликуйте в секции «{recommended_section}» " f"или измените стиль сообщения." } return { "allowed": True, "mode": "active", "sign": sign, "message": "Сообщение совместимо с выбранной секцией." } def _check_compatibility(self, sign: List[float], recommended: str, target: str) -> bool: """Проверяет совместимость сообщения с целевой секцией.""" from config import SECTIONS if target == "general": return True if recommended == target: return True # Проверка по списку разрешённых знаков allowed = SECTIONS.get(target, {}).get("allowed_signs", []) if allowed == "all": return True return sign in allowed def _has_repeated_violations(self, user_id: int, count: int = 3) -> bool: """Проверяет, есть ли у пользователя повторные нарушения.""" # Упрощённая версия — в реальности нужно считать из БД return False # Точка входа для интеграции с API форума def handle_new_post(user_id: int, message: str, section: str): """Функция, вызываемая при новом сообщении на форуме.""" bot = ModeratorBot() result = bot.process_message(user_id, message, section) # Здесь — код отправки результата обратно на форум # (через API, webhook или прямой вызов) return result if __name__ == "__main__": # Тестовый запуск bot = ModeratorBot() test_message = "Можно быть уверенным что трава зелёная." result = bot.process_message(999, test_message, "analytical") print(result)

# requirements.txt flask>=2.0.0 sqlite3 regex>=2021.0.0

# web_interface.py # Простой веб-интерфейс для модераторов from flask import Flask, render_template, request, jsonify from database import Database from moderator_bot import ModeratorBot app = Flask(__name__) db = Database() bot = ModeratorBot() @app.route('/') def index(): return "Семиотический модератор — панель управления" @app.route('/messages/pending') def pending_messages(): """Список сообщений, ожидающих ручной проверки.""" # Здесь — код получения из БД return jsonify({"messages": []}) @app.route('/messages/<int:message_id>/override', methods=['POST']) def override_decision(message_id): """Ручное переопределение решения скрипта.""" new_decision = request.json.get('decision') # Логика переопределения return jsonify({"status": "ok"}) @app.route('/user/<int:user_id>/profile') def user_profile(user_id): """Семиотический профиль пользователя.""" # Здесь — код получения профиля из БД return jsonify({"profile": {}}) if __name__ == '__main__': app.run(debug=True)

Статистика за апрель 2026: - Аналитическая секция: 245 сообщений - Экспрессивная секция: 89 сообщений - Схематическая секция: 67 сообщений - Герметический дискурс: 34 сообщения - Конфликтов при пересечении секций: 12 - Апелляций удовлетворено: 3 из 8