quatuorbellefeuille.com

dump.py (11000B)

---

 #!/usr/bin/env python3
 
 from collections import Counter, defaultdict
 import csv
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from enum import Enum
 from logging import warning
 from pathlib import Path
 import re
 from statistics import mean, median, stdev
 from sys import argv
 from urllib.parse import unquote, urlparse
 from typing import Dict, List, Tuple
 
 import user_agents
 
 
 ACCESS_RE = re.compile(' '.join((
     r'(?P<address>\S+)',
     r'\S+',
     r'\S+',
     r'(?P<date>\[\S+ \S+\])',
     r'"GET (?P<resource>[^ ?]+)(\?\S+)? [^"]+"',
     r'200 [0-9]+',
     r'"(?P<referer>[^"]+)(\?\S+)?"',
     r'"(?P<useragent>[^"]+)"'
 )))
 
 DATE_FMT = '[%d/%b/%Y:%H:%M:%S %z]'
 
 VISIT_MAX_DURATION = timedelta(hours=1)
 
 DOMAINS = (
     'quatuorbellefeuille.com',
     'quatuorbellefeuille.fr',
     'klg.uber.space'
 )
 
 
 def normalize_path(p):
     if p in ('', '/'):
         return '/index.html'
     return unquote(p)
 
 
 @dataclass
 class Access:
     address: str
     useragent: str
     referrer: str
     time: datetime
     resource: str
 
     @classmethod
     def from_log(cls, info):
         resource = normalize_path(info['resource'])
 
         referrer = urlparse(info['referer'])
         if referrer.netloc.endswith(DOMAINS):
             ref = normalize_path(referrer.path)
         elif referrer.netloc:
             ref = referrer.netloc
         else:
             ref = 'n/a'
 
         return cls(
             info['address'], info['useragent'], ref,
             datetime.strptime(info['date'], DATE_FMT), resource
         )
 
 
 def interesting(resource):
     return resource.endswith('.html') or resource == '/'
 
 
 def parse(logs_paths):
     logs = []
     for lp in logs_paths:
         with open(lp) as logs_file:
             logs += logs_file.read().splitlines()
 
     matches = (ACCESS_RE.match(l) for l in logs)
     return tuple(
         Access.from_log(m) for m in matches
         if (m is not None and interesting(m['resource']))
     )
 
 
 Visit = List[Access]
 
 
 class UserAgentKind(Enum):
     PC = 'pc'
     MOBILE = 'mobile'
     TABLET = 'tablet'
     BOT = 'bot'
     NA = 'n/a'
 
     @classmethod
     def from_useragent(cls, ua_string):
         ua = user_agents.parse(ua_string)
         # is_bot is not mutually exclusive with other is_* predicates.
         if ua.is_bot:
             return cls.BOT
         if ua.is_pc:
             return cls.PC
         if ua.is_mobile:
             return cls.MOBILE
         if ua.is_tablet:
             return cls.TABLET
         warning(f'Unknown user agent kind: {ua_string}')
         return cls.NA
 
     def is_human(self):
         return self in {
             UserAgentKind.PC,
             UserAgentKind.MOBILE,
             UserAgentKind.TABLET
         }
 
 
 @dataclass
 class Visitor:
     address: str
     useragent: UserAgentKind
     visits: List[Visit]
 
 
 def sort_visits(accesses):
     visitors: Dict[Tuple(str, str), Visitor] = {}
 
     for a in accesses:
         key = (a.address, a.useragent)
 
         visitor = visitors.get(key)
         if visitor is None:
             visitor = Visitor(
                 a.address,
                 UserAgentKind.from_useragent(a.useragent),
                 [[a]]
             )
             visitors[key] = visitor
             continue
 
         last_visit = visitor.visits[-1]
         last_access = last_visit[-1].time
         if a.time - last_access < VISIT_MAX_DURATION:
             last_visit.append(a)
             continue
 
         visitor.visits.append([a])
 
     return visitors
 
 
 def datetime_day(dt):
     return dt.replace(hour=0, minute=0, second=0)
 
 
 def find_days(visits):
     return {
         datetime_day(v[0].time)
         for v in visits
     }
 
 
 def find_pages(visitors):
     return sorted({
         access.resource
         for v in visitors
         for visit in v.visits
         for access in visit
         if v.useragent.is_human()
     })
 
 
 def external_referrer(ref):
     return ref != 'n/a' and not ref.startswith('/')
 
 
 def simplify_referrer(ref):
     parts = ref.split('.')
     # Remove leading parts (www., l., m.…) and extension (.com, .fr…).
     return parts[-2]
 
 
 def find_referrers(visitors):
     return sorted({
         simplify_referrer(access.referrer)
         for v in visitors
         for visit in v.visits
         for access in visit
         if external_referrer(access.referrer)
     })
 
 
 def daily_visitors(visitors, output_dir):
     days: Dict[datetime, Counter] = defaultdict(Counter)
     columns = ('mobile', 'tablet', 'pc', 'bot', 'n/a')
 
     print('Visitors:')
 
     for v in visitors.values():
         for day in find_days(v.visits):
             days[day][v.useragent.value] += 1
 
     with open(Path(output_dir).joinpath('dailyvisitors.csv'), 'w') as f:
         out = csv.writer(f)
         out.writerow(('day', 'total', *columns))
         print('day', 'total', *columns, sep='\t')
 
         for day in sorted(days):
             counter = days[day]
             counters = tuple(counter[c] for c in columns)
             values = (day.strftime('%F'), sum(counters), *counters)
 
             out.writerow(values)
             print(*values, sep='\t')
 
     return days
 
 
 def daily_visits(visitors, output_dir):
     days: Dict[datetime, Counter] = defaultdict(Counter)
     columns = ('mobile', 'tablet', 'pc', 'bot', 'n/a')
 
     print('Visits:')
 
     for v in visitors.values():
         for visit in v.visits:
             day = datetime_day(visit[0].time)
             days[day][v.useragent.value] += 1
 
     with open(Path(output_dir, 'dailyvisits.csv'), 'w') as f:
         out = csv.writer(f)
         out.writerow(('day', 'total', *columns))
         print('day', 'total', *columns, sep='\t')
 
         for day in sorted(days):
             counter = days[day]
             counters = tuple(counter[c] for c in columns)
             values = (day.strftime('%F'), sum(counters), *counters)
 
             out.writerow(values)
             print(*values, sep='\t')
 
     return days
 
 
 def daily_pages_per_visit(visitors, output_dir):
     days: Dict[datetime, list] = defaultdict(list)
     columns = ('min', 'max', 'med', 'avg', 'dev')
 
     print('Pages/visit:')
 
     for v in visitors.values():
         if not v.useragent.is_human():
             continue
 
         for visit in v.visits:
             day = datetime_day(visit[0].time)
             days[day].append(len(visit))
 
     with open(Path(output_dir, 'dailypagespervisit.csv'), 'w') as f:
         out = csv.writer(f)
         out.writerow(('day', *columns))
         print('day', *columns, sep='\t')
 
         for day in sorted(days):
             view_counts = days[day]
             values = (
                 day.strftime('%F'),
                 min(view_counts),
                 max(view_counts),
                 median(view_counts),
                 mean(view_counts),
                 stdev(view_counts)
             )
 
             out.writerow(values)
             print(*values[:4], *(f'{v:.2f}' for v in values[4:]), sep='\t')
 
     return days
 
 
 def daily_page_hits(visitors, output_dir):
     days: Dict[datetime, Counter] = defaultdict(Counter)
     columns = find_pages(visitors.values())
 
     print('Page hits:')
 
     for v in visitors.values():
         if not v.useragent.is_human():
             continue
 
         for visit in v.visits:
             day = datetime_day(visit[0].time)
             for access in visit:
                 days[day][access.resource] += 1
 
     with open(Path(output_dir, 'dailypagehits.csv'), 'w') as f:
         out = csv.writer(f)
         out.writerow(('day', *columns))
 
         for day in sorted(days):
             page_hits = days[day]
             values = (day.strftime('%F'),
                       *(page_hits[page] for page in columns))
             out.writerow(values)
 
             print(day.strftime('%F'))
             for page, hits in page_hits.most_common(5):
                 print(hits, page, sep='\t')
 
     return days
 
 
 def daily_referrers(visitors, output_dir):
     days: Dict[datetime, Counter] = defaultdict(Counter)
     columns = find_referrers(visitors.values())
 
     print('Referrers:')
 
     for v in visitors.values():
         for visit in v.visits:
             day = datetime_day(visit[0].time)
 
             for access in visit:
                 if not external_referrer(access.referrer):
                     continue
                 days[day][simplify_referrer(access.referrer)] += 1
 
     with open(Path(output_dir, 'dailyreferrers.csv'), 'w') as f:
         out = csv.writer(f)
         out.writerow(('day', *columns))
         print('day', *columns, sep='\t')
 
         for day in sorted(days):
             refcounts = days[day]
             values = (day.strftime('%F'), *(refcounts[ref] for ref in columns))
 
             out.writerow(values)
             print(*values, sep='\t')
 
     return days
 
 
 def dump_stats(visitors, output_dir):
     output_dir = Path(output_dir)
     daily_visitors(visitors, output_dir)
     visits_pday = daily_visits(visitors, output_dir)
     pagespervisit_pday = daily_pages_per_visit(visitors, output_dir)
     pagehits_pday = daily_page_hits(visitors, output_dir)
     referrers_pday = daily_referrers(visitors, output_dir)
 
     ua_values = tuple(ua.value for ua in UserAgentKind)
 
     nb_visitors = {
         ua.value: sum(1 for v in visitors.values() if v.useragent == ua)
         for ua in UserAgentKind
     }
     nb_visits = {
         ua: sum(visits_pday[day][ua] for day in visits_pday)
         for ua in ua_values
     }
     pages_per_visit = tuple(
         nb for day in pagespervisit_pday for nb in pagespervisit_pday[day]
     )
     hits_per_page = {
         page: sum(pagehits_pday[day][page] for day in pagehits_pday)
         for page in find_pages(visitors.values())
     }
     referrers = {
         ref: sum(referrers_pday[day][ref] for day in referrers_pday)
         for ref in find_referrers(visitors.values())
     }
 
     with open(Path(output_dir, 'global.csv'), 'w') as f:
         out = csv.writer(f)
         out.writerow(('#visitors',))
         out.writerows(
             (ua, nb_visitors[ua]) for ua in ua_values
         )
         out.writerow(('total', sum(nb_visitors.values())))
 
         out.writerow(('#visits',))
         out.writerows(
             (ua, nb_visits[ua]) for ua in ua_values
         )
         out.writerow(('total', sum(nb_visits.values())))
 
         out.writerow(('#pages/visit',))
         out.writerows((
             ('min', min(pages_per_visit)),
             ('max', max(pages_per_visit)),
             ('med', median(pages_per_visit)),
             ('avg', mean(pages_per_visit)),
             ('dev', stdev(pages_per_visit))
         ))
 
         out.writerow(('#views/page',))
         lines = reversed(sorted(hits_per_page.items(), key=lambda kv: kv[1]))
         out.writerows(lines)
 
         out.writerow(('#referrers',))
         lines = reversed(sorted(referrers.items(), key=lambda kv: kv[1]))
         out.writerows(lines)
 
 
 def main(logs_paths, output_dir):
     accesses = parse(logs_paths)
     visitors = sort_visits(accesses)
     dump_stats(visitors, output_dir)
 
 
 if __name__ == '__main__':
     main(argv[1:-1], argv[-1])