gaokao-site/data_check.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""GaoKao data quality check script"""

import sqlite3
from collections import defaultdict
from datetime import datetime

DB_PATH = '/www/wwwroot/gaokao/gaokao_henan.db'

def conn():
    c = sqlite3.connect(DB_PATH)
    c.row_factory = sqlite3.Row
    return c

def hdr(t):
    print('=' * 70)
    print('  [' + t + ']')
    print('=' * 70)

def ftr(m):
    print('-' * 70)
    print('  ' + m)
    print()

def check_duplicates(c):
    hdr('Check 1: Fully duplicate rows')
    cols = ['year','school_name','subject','batch','plan_type','subject_req','plan_count','min_score','min_rank','batch_diff']
    ks = ', '.join(cols)
    cur = c.execute('SELECT ' + ks + ', COUNT(*) as cnt, GROUP_CONCAT(id) as ids FROM schools GROUP BY ' + ks + ' HAVING cnt > 1 ORDER BY cnt DESC')
    rows = cur.fetchall()
    if not rows:
        print('  No duplicates found.')
        ftr('PASS')
        return
    td = sum(r['cnt'] - 1 for r in rows)
    print('  Found ' + str(len(rows)) + ' groups, ' + str(td) + ' duplicate rows.')
    for r in rows[:15]:
        print('    x' + str(r['cnt']) + ' | IDs: ' + str(r['ids']))
        print('      ' + str(r['year']) + ' | ' + r['school_name'] + ' | ' + r['subject'] + ' | ' + r['batch'])
        print('      score=' + str(r['min_score']) + ' rank=' + str(r['min_rank']))
    if len(rows) > 15:
        print('  ... and ' + str(len(rows)-15) + ' more groups')
    ftr('Total duplicates: ' + str(td))

def check_empty(c):
    hdr('Check 2: NULL min_score or min_rank')
    t = c.execute('SELECT COUNT(*) as c FROM schools').fetchone()['c']
    es = c.execute('SELECT COUNT(*) as c FROM schools WHERE min_score IS NULL').fetchone()['c']
    er = c.execute('SELECT COUNT(*) as c FROM schools WHERE min_rank IS NULL').fetchone()['c']
    eb = c.execute('SELECT COUNT(*) as c FROM schools WHERE min_score IS NULL AND min_rank IS NULL').fetchone()['c']
    print('  Total records: ' + str(t))
    print('  NULL score: ' + str(es))
    print('  NULL rank:  ' + str(er))
    print('  Both NULL:  ' + str(eb))
    if es == 0 and er == 0:
        ftr('PASS')
    else:
        ftr('WARNING: ' + str(es) + ' missing score, ' + str(er) + ' missing rank')

def check_counts(c):
    hdr('Check 3: School counts by year/subject')
    cur = c.execute('SELECT year, subject, COUNT(DISTINCT school_name) as sc, COUNT(*) as rc FROM schools GROUP BY year, subject ORDER BY year DESC, subject')
    rows = cur.fetchall()
    if not rows:
        print('  Empty table.')
        ftr('PASS')
        return
    print('  {0:>6} {1:<12} {2:>8} {3:>8}'.format('Year','Subject','Schools','Records'))
    print('  ' + '-'*6 + ' ' + '-'*12 + ' ' + '-'*8 + ' ' + '-'*8)
    for r in rows:
        print('  {0:>6} {1:<12} {2:>8} {3:>8}'.format(r['year'],r['subject'],r['sc'],r['rc']))
    ftr(str(len(rows)) + ' groups')

def check_anomaly(c):
    hdr('Check 4: Anomalous score diff (same school/year/subject)')
    thr = 80
    cur = c.execute('SELECT year,school_name,subject,MIN(min_score) as mn,MAX(min_score) as mx,AVG(min_score) as av,COUNT(*) as cnt FROM schools WHERE min_score IS NOT NULL GROUP BY year,school_name,subject HAVING (mx-mn) > ? ORDER BY (mx-mn) DESC', (thr,))
    rows = cur.fetchall()
    if not rows:
        print('  None found (diff > ' + str(thr) + ').')
        ftr('PASS')
        return
    print('  Groups with diff > ' + str(thr) + ':')
    print('  {0:>6} {1:<20} {2:<10} {3:>6} {4:>6} {5:>8} {6:>6} {7:>6}'.format('Year','School','Subject','Min','Max','Avg','Cnt','Diff'))
    print('  ' + '-'*6 + ' ' + '-'*20 + ' ' + '-'*10 + ' ' + '-'*6 + ' ' + '-'*6 + ' ' + '-'*8 + ' ' + '-'*6 + ' ' + '-'*6)
    for r in rows[:20]:
        d = r['mx'] - r['mn']
        print('  {0:>6} {1:<20} {2:<10} {3:>6} {4:>6} {5:>8.1f} {6:>6} {7:>6}'.format(r['year'],str(r['school_name'])[:20],r['subject'],r['mn'],r['mx'],r['av'],r['cnt'],d))
    if len(rows) > 20:
        print('  ... and ' + str(len(rows)-20) + ' more')
    for r in rows[:5]:
        print()
        print('  ' + str(r['year']) + ' ' + r['school_name'] + ' ' + r['subject'] + ' diff=' + str(r['mx']-r['mn']))
        dd = c.execute('SELECT id,batch,plan_type,subject_req,plan_count,min_score,min_rank FROM schools WHERE year=? AND school_name=? AND subject=? AND min_score IS NOT NULL ORDER BY min_score', (r['year'],r['school_name'],r['subject']))
        for d in dd:
            print('    ID=' + str(d['id']) + ' ' + str(d['batch']) + ' ' + str(d['plan_type']) + ' score=' + str(d['min_score']) + ' rank=' + str(d['min_rank']))
    ftr(str(len(rows)) + ' groups to review')

def check_missing_rank(c):
    hdr('Check 5: Schools with missing rank')
    cur = c.execute('SELECT year,school_name,subject,batch,plan_type,min_score FROM schools WHERE min_rank IS NULL ORDER BY year DESC, school_name')
    rows = cur.fetchall()
    if not rows:
        print('  All records have rank.')
        ftr('PASS')
        return
    bs = defaultdict(int)
    for r in rows:
        bs[(r['year'], r['school_name'])] += 1
    print('  ' + str(len(rows)) + ' records, ' + str(len(bs)) + ' schools affected.')
    print('  {0:>6} {1:<20} {2:>10}'.format('Year','School','Missing'))
    print('  ' + '-'*6 + ' ' + '-'*20 + ' ' + '-'*10)
    for (y,n),c in sorted(bs.items(),key=lambda x:-x[1])[:20]:
        print('  {0:>6} {1:<20} {2:>10}'.format(y,n,c))
    if len(bs) > 20:
        print('  ... and ' + str(len(bs)-20) + ' more schools')
    for r in rows[:10]:
        print('    ' + str(r['year']) + ' ' + r['school_name'] + ' ' + r['subject'] + ' ' + r['batch'] + ' score=' + str(r['min_score']))
    ftr(str(len(rows)) + ' records need rank data')

def main():
    print()
    print('=' * 70)
    print('  Gaokao Henan Data Quality Check')
    print('  DB: gaokao_henan.db')
    print('=' * 70)
    print('  Time: ' + datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    print('  Path: ' + DB_PATH)
    print()
    db = conn()
    try:
        check_duplicates(db)
        check_empty(db)
        check_counts(db)
        check_anomaly(db)
        check_missing_rank(db)
    finally:
        db.close()
    print('=' * 70)
    print('  Check Complete.')
    print('=' * 70)
    print()

if __name__ == '__main__':
    main()