json 接口对比 python

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import os
import urllib
import traceback
import json
import logging
import types
import re
import csv

logging.basicConfig(level=logging.DEBUG,
    format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
    datefmt='%a, %d %b %Y %H:%M:%S',
    filename='check.log',
    filemode='w+')

logger = logging.getLogger(__name__)

def load_url(url_ifn):
    list_url = []
    with open(url_ifn) as url_file:
        for line in url_file:
            line = unicode(line.strip(), "utf-8", "ignore")
            list_url.append(line)
    return list_url


def curl_url(url):
    content = u''  
    try:   
        status = urllib.urlopen(url)   
        if status.getcode() == 200:   
            content = status.read()   
            content = unicode(content.strip(), 'utf-8', 'ignore')   
        else:   
            logger.error('fetch error [%s]' % url)   
    except:   
        logger.error('fetch error %s' % traceback.format_exc())
    return content  


def check_dict(dict_old, dict_new, url, ofh):
    global global_check_res
    set_old = set(dict_old.keys())
    set_new = set(dict_new.keys())

    set_same = set_old & set_new
    set_only_old = set_old - set_same
    set_only_new = set_new - set_same
    ofh.write("only_old:[%s] only_new:[%s]\n same:[%s]" % (",".join(set_only_old), ",".join(set_only_new), ",".join(set_same)))
    global_check_res = False


def check_segmentlist(seg_old, seg_new, url, ofh, from_msg):
    global global_check_res
    str_old = '->'.join([item['busid'] for item in seg_old if 'busid' in item])
    str_new = '->'.join([item['busid'] for item in seg_new if 'busid' in item])
    if str_old != str_new:
        ofh.write("[%s] segment_list old:[%s] new[%s]\n" % (from_msg, str_old, str_new))
        global_check_res = False
        return False
    else:
        return True
        

def check_json(old_obj, new_obj, url, ofh, from_msg):
    global global_check_res
    if type(old_obj) != type(new_obj):
        global_check_res = False      
        return False
    obj_type = type(old_obj)
    if obj_type == types.DictType:
        set_delete = set(["code", "version", "timestamp"])
        for key in old_obj.keys():
            if key in set_delete:
                continue
            if key == "color":
                old_obj[key] = old_obj[key].lower()
                new_obj[key] = new_obj[key].lower()
            if key not in new_obj:
                ofh.write("[%s]\t[%s]new not exist\n" % (from_msg, key))
                global_check_res = False
                continue
            if key == "segmentlist":
                if not check_segmentlist(old_obj[key], new_obj[key], url, ofh, "%s->%s" % (from_msg, key)):
                    continue
            if check_json(old_obj[key], new_obj[key], url, ofh, "%s->%s" % (from_msg, key)):
                continue
            else:
                ofh.write("[%s]\t[%s]\n" % (from_msg, key))
                global_check_res = False
        for key in new_obj.keys():
            if key not in old_obj:
                ofh.write("[%s]\t[%s]old not exist\n" % (from_msg, key))
                global_check_res = False

    elif obj_type == types.ListType:
        if len(old_obj) != len(new_obj):
            ofh.write("[%s]\tarray len not equals old[%d] new[%d]\n" % (from_msg, len(old_obj), len(new_obj))) 
            global_check_res = False

        for i in range(min(len(old_obj), len(new_obj))):
            if check_json(old_obj[i], new_obj[i], url, ofh, "%s->at[%s]" % (from_msg, i)):
                continue
            else:
                ofh.write("[%s]\tarray at [%s] \n" % (from_msg, i))			
                global_check_res = False
    else:
        if old_obj == new_obj:
            return True
        else:
            try:
                ofh.write("[%s]\t old:[%s] new:[%s]\n" % (from_msg, str(old_obj).encode("utf-8"), str(new_obj).encode("utf-8")))			
                global_check_res = False
            except:
                ofh.write("[%s] value not equal\n" % (from_msg))
                global_check_res = False
            return False
    return True


def main(url_ifn, check_ofn, old_url_prefix, new_url_prefix):
    global global_check_res
    list_url = load_url(url_ifn)
    print "load url line %d" % len(list_url)

    list_real_url = []
    for url in list_url:
        pars = url.split("?")
        if len(pars) < 2:
            print "fail:[%s]" % url
            continue
        list_real_url.append(pars[1])
    print "load real url line %d" % len(list_real_url)
    
    check_ofh = open(check_ofn, "w+")
    success_count = 0
    total_count = 0
    for i in range(len(list_real_url)):
        try:
            if i%100 == 0:
                logger.error("curl len[%d]" % i)
            url = list_real_url[i]

            old_url = "%s%s" % (old_url_prefix, url)
            new_url = "%s%s" % (new_url_prefix, url)
            
            old_content = curl_url(old_url)
            new_content = curl_url(new_url)
            
            if not old_content or not new_content:
                logger.error("[%s][content is null]" % url)
                continue

            try:
                old_obj = json.loads(old_content)
            except:
                logger.error("[%s] error [json parse error]" % (old_url))
                continue

            try:
                new_obj = json.loads(new_content)
            except:
                logger.error("[%s] error [json parse error]" % (new_url))
                continue

            global_check_res = True
            total_count += 1

            check_ofh.write("%s\n" % old_url)
            check_ofh.write("%s\n" % new_url)
            check_json(old_obj, new_obj, url, check_ofh, "parent")

            if global_check_res:
                success_count += 1
                check_ofh.write("[%s]\t[success]\n" % url)
            else:
                writer.writerow([old_url])
        except:
            logger.error("[%s] error [%s]" % (url, traceback.format_exc()))
    check_ofh.write("---------------------------\n")
    check_ofh.write("total_count:%d\n" %  total_count)
    check_ofh.write("success_count:%s\n" % success_count)
    check_ofh.write("fail_count:%s\n" % (total_count - success_count))
    check_ofh.close()

global_check_res = True
if __name__ == '__main__':
    url_ifn = ""
    if len(sys.argv) <= 1:
        sys.exit(1)
    url_ifn = sys.argv[1]
    check_ofn = "./check_res.txt"
    csvfile = file('diff.csv', 'w')
    writer = csv.writer(csvfile, dialect='excel')

    old_url_prefix = "http://127.0.0.1:14669/busservice?"
    new_url_prefix = "http://127.0.0.1:14668/busservice?"
    main(url_ifn, check_ofn, old_url_prefix, new_url_prefix)