#!/usr/bin/env python3 """Просмотры публикаций по месяцам: из журнала замеров в сводный отчёт. Зачем. До 14.09.2026 просмотры хранились одним числом на публикацию, и новый замер затирал старый. Показать, сколько статья набрала в сентябре, а сколько в октябре, было не из чего. Теперь замеры копятся в `завод/просмотры-журнал.csv` (одна строка - один замер), а этот скрипт раскладывает их по месяцам. Как считаем. Счётчики площадок показывают сумму за всё время, поэтому прирост за месяц - это разница между последним замером месяца и последним замером предыдущего. Если в месяце не мерили, значение переносится с прошлого, и прирост выходит нулевым: это честно, роста мы не видели. Отсчёт с сентября 2026: портал поднят 02.09.2026, раньше публикаций не было (решение руководителя 14.09.2026). Порядок: после обновить-статьи.py, до обновить-фолбэк.py. Запуск: python3 панель/данные/обновить-просмотры.py """ import json import re from datetime import date from pathlib import Path ПАПКА = Path(__file__).resolve().parent КОРЕНЬ = ПАПКА.parent.parent ЖУРНАЛ = КОРЕНЬ / "завод" / "просмотры-журнал.csv" РЕЕСТР = КОРЕНЬ / "завод" / "реестр-текстов.json" ОТЧЁТ = ПАПКА / "report.json" ПОСЛЕДНИЕ = КОРЕНЬ / "завод" / "просмотры-публикаций.csv" НАЧАЛО = (2026, 9) # сентябрь 2026, раньше публикаций не было МЕСЯЦЫ = ("января", "февраля", "марта", "апреля", "мая", "июня", "июля", "августа", "сентября", "октября", "ноября", "декабря") КОРОТКО = ("янв", "фев", "мар", "апр", "май", "июн", "июл", "авг", "сен", "окт", "ноя", "дек") ДАТА_В_СТАТУСЕ = re.compile(r"(\d{2})\.(\d{2})\.(\d{4})") ПОЛНЫЕ = ("Январь", "Февраль", "Март", "Апрель", "Май", "Июнь", "Июль", "Август", "Сентябрь", "Октябрь", "Ноябрь", "Декабрь") def месяцы_до(сегодня): """Список (год, месяц) от начала отсчёта до текущего месяца включительно.""" ряд, г, м = [], *НАЧАЛО while (г, м) <= (сегодня.year, сегодня.month): ряд.append((г, м)) г, м = (г + 1, 1) if м == 12 else (г, м + 1) return ряд СТАТЬИ = ПАПКА / "articles.json" def номера_по_заголовку(): """{(площадка, начало заголовка): №} - чтобы найти статью, когда в журнале не проставлен номер. Номера живут во вкладке «Статьи», в реестре их нет.""" if not СТАТЬИ.exists(): return {} д = json.loads(СТАТЬИ.read_text(encoding="utf-8")) карта = {} for с in д.get("строки", []): н, з = с.get("№"), (с.get("заголовок") or "").strip().lower() пл = (с.get("площадка") or "").split("(")[0].strip().lower() if н and з: карта[(пл, з[:50])] = н return карта def замеры(реестр=None): """{№: [(дата, просмотры|None, примечание)]} из журнала, по возрастанию даты. Номер статьи во втором поле раньше был обязателен, и строка без него молча пропадала. Так потерялись семь замеров - в том числе первый замер ВКонтакте и Дзена (21.09.2026). Теперь номер ищем по площадке и заголовку, а что не нашлось - называем вслух, а не глотаем. """ по_статьям = {} по_заголовку = номера_по_заголовку() непонятные = [] if not ЖУРНАЛ.exists(): return по_статьям for строка in ЖУРНАЛ.read_text(encoding="utf-8-sig").splitlines(): строка = строка.strip() if not строка or строка.startswith("#") or строка.startswith("дата;"): continue ч = (строка.split(";") + [""] * 6)[:6] дата, номер, площадка, заголовок, просмотры, нота = [э.strip() for э in ч] if not номер.isdigit(): найден = по_заголовку.get((площадка.split("(")[0].strip().lower(), заголовок.strip().lower()[:50])) if not найден: непонятные.append(f"{дата} {площадка}: {заголовок[:45]}") continue номер = str(найден) число = None if просмотры: цифры = re.sub(r"[^\d]", "", просмотры) число = int(цифры) if цифры else None по_статьям.setdefault(int(номер), {"площадка": площадка, "заголовок": заголовок, "замеры": []}) по_статьям[int(номер)]["замеры"].append((дата, число, нота)) for д in по_статьям.values(): д["замеры"].sort(key=lambda з: з[0]) if непонятные: print(" замеры без номера и без совпадения по заголовку (пропущены):") for с in непонятные: print(" ", с) return по_статьям def дата_публикации(реестр, площадка, заголовок): """Дата выхода из статуса материала: «опубликовано на Audit-it 07.09.2026». ГРАБЛИ 16.09.2026: раньше совпадение искалось «по площадке ИЛИ по заголовку», и для второго материала площадки бралась дата ПЕРВОГО. В отчёте сегодняшние публикации Smart-Lab и Spark стояли с датами 3 и 4 сентября, из-за чего попадали в чужую когорту и искажали возраст публикаций. Теперь сначала ищем по заголовку, а площадка - только запасной путь, и то лишь когда у неё в реестре ровно один вышедший материал. """ if заголовок: for т in реестр["тексты"]: if т.get("заголовок", "").startswith(заголовок[:30]): м = ДАТА_В_СТАТУСЕ.search(т.get("статус") or "") if м: д, мес, г = м.groups() return f"{г}-{мес}-{д}" свои = [т for т in реестр["тексты"] if т.get("площадка") == площадка and ДАТА_В_СТАТУСЕ.search(т.get("статус") or "")] if len(свои) == 1: д, мес, г = ДАТА_В_СТАТУСЕ.search(свои[0]["статус"]).groups() return f"{г}-{мес}-{д}" return None def когорта(дата_iso): """Декада выхода: «1-10 сентября 2026». Так публикации одного возраста рядом.""" if not дата_iso: return "дата выхода не записана" г, м, д = (int(э) for э in дата_iso.split("-")) if д <= 10: диапазон = "1-10" elif д <= 20: диапазон = "11-20" else: диапазон = "21-конец" return f"{диапазон} {МЕСЯЦЫ[м - 1]} {г}" def имя_базы(площадка): """«DTF (dtf.ru)» -> «DTF»: в журнале имена с адресом, в панели без.""" return (площадка or "").split("(")[0].strip() or "без площадки" def ряды_площадок(по_статьям): """Накопительные просмотры каждой площадки по датам замеров. Зачем. Карточки «Динамика просмотров по площадкам» и «Рост и застой» в PR-обзоре рисовались только в демо-режиме: на живых данных они писали «первого замера ещё не было» при двадцати публикациях и пятнадцати замерах (найдено 15.09.2026). Ловушка, из-за которой нельзя просто сложить строки журнала одной даты: в замер попадают не все публикации сразу. Если 16-го мерили три статьи из пяти, прямая сумма покажет падение площадки. Поэтому у каждой публикации переносим последнее известное число вперёд. """ даты = sorted({д for з in по_статьям.values() for д, ч, _ in з["замеры"] if ч is not None}) if len(даты) < 2: return None по_площадкам = {} for номер, з in по_статьям.items(): имя = имя_базы(з["площадка"]) известные = {д: ч for д, ч, _ in з["замеры"] if ч is not None} if not известные: continue ряд = по_площадкам.setdefault(имя, {"ряд": [0] * len(даты), "публикаций": 0}) ряд["публикаций"] += 1 посл = 0 for и, д in enumerate(даты): if д in известные: посл = известные[д] ряд["ряд"][и] += посл строки = [{"площадка": имя, "ряд": з["ряд"], "публикаций": з["публикаций"], "всего": з["ряд"][-1]} for имя, з in по_площадкам.items()] строки.sort(key=lambda с: -с["всего"]) return { "даты": даты, "подписи": [".".join(reversed(д.split("-")[1:])) for д in даты], "строки": строки, "подпись": ("Без замера в этот день берём последнее число. Скачок - " "новая публикация."), } def рост_и_застой(по_статьям): """Что выросло и что стоит между двумя последними замерами. Считаем только публикации, которые есть в обоих замерах: новая статья иначе засчиталась бы площадке как рост на весь свой счётчик. """ даты = sorted({д for з in по_статьям.values() for д, ч, _ in з["замеры"] if ч is not None}) if len(даты) < 2: return None было_дата, стало_дата = даты[-2], даты[-1] по_площадкам, новые, не_мерили = {}, [], {} новых_у_площадки = {} # сколько публикаций площадки вышло после прошлого замера for номер, з in по_статьям.items(): имя = имя_базы(з["площадка"]) известные = {д: ч for д, ч, _ in з["замеры"] if ч is not None} было = последнее_до(известные, было_дата) стало = известные.get(стало_дата) if стало is None: # Площадку в свежий замер не включили. Раньше такие строки просто # пропадали, и блок выглядел так, будто у площадки нет данных. не_мерили.setdefault(имя, {"площадка": имя, "последнее": 0, "публикаций": 0, "когда": последняя_дата(известные)}) не_мерили[имя]["публикаций"] += 1 не_мерили[имя]["последнее"] = (не_мерили[имя]["последнее"] or 0) + (было or 0) continue if было is None: новые.append({"площадка": имя, "заголовок": з["заголовок"], "просмотры": стало}) новых_у_площадки[имя] = новых_у_площадки.get(имя, 0) + 1 continue п = по_площадкам.setdefault(имя, {"площадка": имя, "было": 0, "стало": 0, "публикаций": 0}) п["было"] += было п["стало"] += стало п["публикаций"] += 1 строки = [] for п in по_площадкам.values(): п["прирост"] = п["стало"] - п["было"] # Сколько публикаций у площадки всего и сколько попало в сравнение. # 23.09.2026 у Pikabu стояло «публикаций: 3» при четырёх вышедших: # четвёртая вышла в день замера, сравнивать её было не с чем, и # площадка выглядела застоявшейся, хотя новый пост набрал 2000. п["новых"] = новых_у_площадки.get(п["площадка"], 0) п["всего_публикаций"] = п["публикаций"] + п["новых"] строки.append(п) пропущены = [с for с in не_мерили.values() if с["когда"]] без_счётчика = [с for с in не_мерили.values() if not с["когда"]] оговорка = (f"Новых: {len(новые)}, в рост не идут." if новые else "Новых публикаций нет.") if пропущены: оговорка += f" Не мерили: {len(пропущены)}." if без_счётчика: оговорка += f" Без счётчика: {len(без_счётчика)}." растут = sorted([с for с in строки if с["прирост"] > 0], key=lambda с: -с["прирост"]) стоят = sorted([с for с in строки if с["прирост"] <= 0], key=lambda с: -с["стало"]) return { "было_дата": было_дата, "стало_дата": стало_дата, "растут": растут, "стоят": стоят, "новые": новые, # площадки без единого замера - это не «не мерили в этот раз», # у них просто нет счётчика; держим их отдельной строкой "не_мерили": sorted([с for с in не_мерили.values() if с["когда"]], key=lambda с: -(с["последнее"] or 0)), "без_счётчика": sorted(с["площадка"] for с in не_мерили.values() if not с["когда"]), "подпись": (f"Замер {по_русски_дата(стало_дата)} против " f"{по_русски_дата(было_дата)}, только публикации с обоими замерами."), "оговорка": оговорка, } def последняя_дата(известные): return max(известные) if известные else None def последнее_до(известные, дата): свои = [д for д in известные if д <= дата] return известные[max(свои)] if свои else None def по_русски_дата(д): return ".".join(reversed(д.split("-"))) def свод_когорт(строки, подписи): """Публикации, сгруппированные по декаде выхода. Зачем. В панели срез «Когортный анализ» у контура PR показывал табличку «когорты у внешнего контура не считаются: ни одна площадка не отдаёт просмотры по датам». Это перестало быть правдой, когда мы завели свой журнал замеров: даты у нас теперь есть, и когорта каждой публикации уже лежит в строке (найдено 15.09.2026). Считаем среднее на публикацию, а не только сумму: когорты разного размера, и когорта из четырнадцати публикаций будет выглядеть успешнее пятёрки просто потому, что её больше. """ по_когортам = {} for с in строки: по_когортам.setdefault(с["когорта"], []).append(с) итог = [] for имя, свои in по_когортам.items(): измеренные = [с for с in свои if с["итого"] is not None] сумма = sum(с["итого"] for с in измеренные) помесячно = [0] * len(подписи) for с in измеренные: for и, зн in enumerate(с["по_месяцам"][:len(подписи)]): помесячно[и] += зн or 0 итог.append({ "когорта": имя, "публикаций": len(свои), "с_замером": len(измеренные), "без_замера": len(свои) - len(измеренные), "всего": сумма, "на_публикацию": round(сумма / len(измеренные)) if измеренные else None, "по_месяцам": помесячно, "вышли": sorted(с["опубликована"] for с in свои if с["опубликована"]), }) # от свежей когорты к старой: сверху то, что вышло последним итог.sort(key=lambda к: (к["вышли"][-1] if к["вышли"] else ""), reverse=True) без = sum(к["без_замера"] for к in итог) return { "строки": итог, "месяцы": подписи, "подпись": ("Когорта - декада выхода: публикации одного возраста стоят рядом. " "Среднее на публикацию важнее суммы: когорты разного размера."), "оговорка": (f"Без замера {без} " + склонение_пуб(без) + ": площадка не отдаёт счётчик " "или замер ещё не вписан в журнал. Такие публикации в среднее " "не идут." if без else "Замерены все публикации когорт."), "источник": "завод/просмотры-журнал.csv", } def склонение_пуб(н): н = abs(int(н)) if н % 10 == 1 and н % 100 != 11: return "публикация" if 2 <= н % 10 <= 4 and not 12 <= н % 100 <= 14: return "публикации" return "публикаций" def главная(): сегодня = date.today() ряд = месяцы_до(сегодня) # Полное название месяца, как в колонках эталона PR Factory # («Август 2026», а не «авг 2026»): колонок мало, место есть. подписи = [f"{ПОЛНЫЕ[м - 1]} {г}" for г, м in ряд] реестр = json.loads(РЕЕСТР.read_text(encoding="utf-8")) if РЕЕСТР.exists() else {"тексты": []} отчёт = json.loads(ОТЧЁТ.read_text(encoding="utf-8")) строки = [] for номер, д in sorted(замеры(реестр).items()): # значение счётчика на конец каждого месяца на_конец, последнее = [], None for г, м in ряд: for дата, число, _ in д["замеры"]: if not дата or число is None: continue гд, мд = int(дата[:4]), int(дата[5:7]) if (гд, мд) == (г, м): последнее = число на_конец.append(последнее) # прирост за месяц: разница с прошлым месяцем, первый месяц целиком наш по_месяцам, прошлое = [], 0 for значение in на_конец: if значение is None: по_месяцам.append(None) continue по_месяцам.append(max(0, значение - прошлое)) прошлое = значение измеряли = [з for з in д["замеры"] if з[1] is not None] ноты = [з[2] for з in д["замеры"] if з[2]] дата_выхода = дата_публикации(реестр, д["площадка"], д["заголовок"]) строки.append({ "№": номер, "статья": д["заголовок"], "площадка": д["площадка"], "опубликована": дата_выхода, "когорта": когорта(дата_выхода), "по_месяцам": по_месяцам, "итого": прошлое if измеряли else None, "замеров": len(измеряли), "последний_замер": измеряли[-1][0] if измеряли else None, "нота": ноты[-1] if ноты else ("не проверялась" if not измеряли else ""), "демо": False, }) # Публикации, по которым замеров нет вовсе, раньше в отчёт не попадали: # список строился только из журнала. 16.09.2026 из-за этого отчёт показывал # 19 публикаций вместо 21 - выпали материалы площадок без счётчика # («Работаем!») и свежие, которые ещё не мерили. Добираем их из реестра # с пустыми значениями и честной причиной. # Сопоставляем по номеру публикации из вкладки «Статьи», а не по названию: # площадка в журнале и в реестре пишется по-разному («myJane» против # «myJane (сеть Медиафорт)»), а заголовок статьи после выхода иногда # правится - у Pressfeed замер лежал под старым названием, и материал # попадал в отчёт дважды. таблица = json.loads((ПАПКА / "articles.json").read_text(encoding="utf-8")) номера_статей = {} for с_ in таблица["строки"]: номера_статей[(с_.get("заголовок") or "")[:30]] = с_.get("№") уже = {с["№"] for с in строки} # запас нумерации считаем от ВСЕХ известных номеров, включая вкладку # «Статьи»: иначе новый номер налезает на занятый и в отчёте появляются # две строки с одним № номер_запас = max([с["№"] for с in строки] + [с_.get("№") or 0 for с_ in таблица["строки"]], default=0) for т in реестр["тексты"]: если_вышел = (т["вид"] == "внешняя площадка" and "опубликован" in (т["статус"] or "").lower()) if not если_вышел: continue номер = номера_статей.get((т.get("заголовок") or "")[:30]) if номер in уже: continue номер_запас += 1 дата_выхода = дата_публикации(реестр, т.get("площадка"), т.get("заголовок")) строки.append({ "№": номер or номер_запас, "статья": т.get("заголовок") or "", "площадка": т.get("площадка") or "", "опубликована": дата_выхода, "когорта": когорта(дата_выхода), "по_месяцам": [None] * len(ряд), "итого": None, "замеров": 0, "последний_замер": None, "нота": "счётчика у площадки нет либо замер ещё не делали", "демо": False, }) строки.sort(key=lambda с: (с["опубликована"] or "0000", с["№"]), reverse=True) отчёт.setdefault("pr", {})["по_публикациям"] = { "месяцы": подписи, "строки": строки, "подпись": ("Одна строка - одна публикация на площадке. В колонках месяцев не сумма, " "а прирост за месяц: счётчики площадок считают нарастающим итогом, поэтому " "берём разницу между замерами. Пустая клетка значит «в этом месяце не мерили», " "а не «просмотров не было». Отсчёт с сентября 2026: раньше публикаций не было."), "источник": "завод/просмотры-журнал.csv (замеры вписываются руками из кабинетов площадок)", } отчёт["pr"]["когорты"] = свод_когорт(строки, подписи) журнал = замеры(реестр) динамика = ряды_площадок(журнал) if динамика: отчёт["pr"]["динамика"] = динамика # Подписи карточки раньше лежали в report.json словами и отстали от # содержимого: обещали «прибавку за последний месяц» и «порядок как # в рейтинге выше», когда шаг стал замером (15.09.2026). отчёт["pr"]["динамика_подпись"] = ( f"Шаг - замер: их {len(динамика['даты'])}, последний " f"{по_русски_дата(динамика['даты'][-1])}. Сверху самые читаемые.") рост = рост_и_застой(журнал) if рост: отчёт["pr"]["рост"] = рост без = len(рост.get("без_счётчика") or []) отчёт["pr"]["динамика_оговорка"] = ( "Счётчики площадок считают по-разному: ряды показывают направление, не точные цифры." + (f" У {без} " + ("площадки" if без % 10 == 1 and без % 100 != 11 else "площадок") + " своего счётчика нет вовсе: " + ", ".join(рост["без_счётчика"]) + "." if без else "")) ОТЧЁТ.write_text(json.dumps(отчёт, ensure_ascii=False, indent=2), encoding="utf-8") # Старая таблица «последнее значение» остаётся: по ней считают рейтинг # площадок и колонку просмотров во вкладке «Статьи». Чтобы руками вписывать # замеры в одно место, пересобираем её из журнала, а не ведём отдельно. свежие = [] for с in sorted(строки, key=lambda с: с["№"]): число = "" if с["итого"] is None else с["итого"] дата = с["последний_замер"] or (с["нота"] or "") свежие.append(f'{с["№"]};{с["площадка"]};{с["статья"]};{число};{дата}') шапка = [ "# Последний замер по каждой публикации. ФАЙЛ СОБИРАЕТСЯ САМ из", "# завод/просмотры-журнал.csv - руками не правим, иначе правка пропадёт", "# при следующем прогоне панель/данные/обновить-просмотры.py.", "# Новые замеры вписываем в журнал: там одна строка - один замер.", "№;площадка;заголовок;просмотры;дата", ] ПОСЛЕДНИЕ.write_text("\n".join(шапка + свежие) + "\n", encoding="utf-8") сизмеренных = sum(1 for с in строки if с["итого"] is not None) print(f"публикаций в таблице: {len(строки)}, с замером: {сизмеренных}, " f"месяцев в шкале: {len(подписи)} ({', '.join(подписи)})") if __name__ == "__main__": главная()