manyi/venv/Lib/site-packages/addressparser/__init__.py

# -*- coding: utf-8 -*-
"""
@author:XuMing（xuming624@qq.com)
@description:
"""

import os
import six
import warnings
try:
    from collections.abc import Iterable
except ImportError:
    from collections import Iterable
import pandas as pd

from .structures import AddrMap, Pca
from .structures import P, C, A

__version__ = "0.2.4"

pwd_path = os.path.abspath(os.path.dirname(__file__))
# 区划地址文件
pca_path = os.path.join(pwd_path, 'pca.csv')

if six.PY2:
    text_type = unicode
else:
    text_type = str


def convert_to_unicode(text):
    """Converts `text` to Unicode (if it's not already), assuming utf-8 input."""
    if not isinstance(text, text_type):
        try:
            text = text.decode('utf-8')
        except UnicodeDecodeError:
            text = text.decode('gbk', 'ignore')
        except Exception as e:
            warnings.warn('Convert to unicode error: %s, text: %s' % (e, text))
    return text


def _data_from_csv():
    """
    从csv文件获取数据
    :return: (AddrMap, AddrMap, AddrMap, dict, dict)
    """
    # 区名及其简写 -> 相关pca元组
    area_map = AddrMap()
    # 城市名及其简写 -> 相关pca元组
    city_map = AddrMap()
    # (省名全称, 区名全称) -> 相关pca元组
    province_area_map = AddrMap()
    # 省名 -> 省全名
    province_map = {}
    # (省名, 市名, 区名) -> (纬度,经度)
    latlng = {}
    # 数据约定:国家直辖市的sheng字段为直辖市名称, 省直辖县的city字段为空

    pca_df = pd.read_csv(pca_path, sep=',', header=0, encoding='utf-8')
    pca_df = pca_df.fillna('')
    for record_dict in pca_df.values:
        # latlng[(record_dict['sheng'], record_dict['shi'], record_dict['qu'])] =
        # (record_dict['lat'], record_dict['lng'])
        record_dict = [convert_to_unicode(i) if i is isinstance(i, text_type) else i for i in record_dict]
        latlng[(record_dict[1], record_dict[2], record_dict[3])] = (record_dict[4], record_dict[5])

        _fill_province_map(province_map, record_dict)
        _fill_area_map(area_map, record_dict)
        _fill_city_map(city_map, record_dict)
        _fill_province_area_map(province_area_map, record_dict)

    return area_map, city_map, province_area_map, province_map, latlng


def _fill_province_area_map(province_area_map, record_dict):
    """
    填充省，区
    :param province_area_map: AddrMap
    :param record_dict:
    :return:
    """
    pca_tuple = (record_dict[1], record_dict[2], record_dict[3])
    key = (record_dict[1], record_dict[3])
    # 第三个参数在此处没有意义
    province_area_map.append_relational_addr(key, pca_tuple, P)


# 过滤混淆区名 '河北北戴河富丽小区1号'
filter_area_names = [u'河北区', u'新城区']
# 处理了部分常见自治县简写
short_area_names = {
    u'白沙黎族自治县': u'白沙县',
    u'昌江黎族自治县': u'昌江县',
    u'乐东黎族自治县': u'乐东县',
    u'陵水黎族自治县': u'陵水县',
    u'保亭黎族苗族自治县': u'保亭县',
    u'琼中黎族苗族自治县': u'琼中县',
    u'长阳土家族自治县': u'长阳县',
    u'五峰土家族自治县': u'五峰县',
    u'大通回族土族自治县': u'大通县',
    u'民和回族土族自治县': u'民和县',
    u'互助土族自治县': u'互助县',
    u'化隆回族自治县': u'化隆县',
    u'循化撒拉族自治县': u'循化县',
    u'青龙满族自治县': u'青龙县',
    u'屏边苗族自治县': u'屏边县',
    u'金平苗族瑶族傣族自治县': u'金平县',
    u'河口瑶族自治县': u'河口县',
    u'丰宁满族自治县': u'丰宁县',
    u'宽城满族自治县': u'宽城县',
    u'围场满族蒙古族自治县': u'围场县',
}


def _fill_area_map(area_map, record_dict):
    """
    填充三级区划（区级）地名，包括简称
    :param area_map:  AddrMap, dict
    :param record_dict: dict
    :return: area_map
    """
    area_name = record_dict[3]
    pca_tuple = (record_dict[1], record_dict[2], record_dict[3])
    area_map.append_relational_addr(area_name, pca_tuple, A)
    # 自治县区划简称
    if area_name in short_area_names.keys():
        area_map.append_relational_addr(short_area_names[area_name], pca_tuple, A)
    # 4字区划简称
    elif len(area_name) > 3 and (area_name.endswith(u'新区') or area_name.endswith(u'城区') or area_name.endswith(u'林区')):
        area_map.append_relational_addr(area_name[:-2], pca_tuple, A)
    # 过滤的区划名称
    elif area_name in filter_area_names:
        pass
    # 3字区划简称，'XX区'不简写
    elif len(area_name) > 2 and (area_name.endswith(u'市') or area_name.endswith(u'县')):
        area_map.append_relational_addr(area_name[:-1], pca_tuple, A)


# 过滤混淆市名 eg '吉林省、吉林市的混淆'
filter_city_names = [u'吉林市']


def _fill_city_map(city_map, record_dict):
    """
    填充二级区划（市级）地名，包括简称
    :param city_map: AddrMap, dict
    :param record_dict: dict
    :return: city_map
    """
    city_name = record_dict[2]  # shi
    pca_tuple = (record_dict[1], record_dict[2], record_dict[3])
    city_map.append_relational_addr(city_name, pca_tuple, C)
    # fix 吉林省、吉林市的混淆
    if city_name in filter_city_names:
        pass
    elif city_name.endswith(u'市'):
        city_map.append_relational_addr(city_name[:-1], pca_tuple, C)
    # 特别行政区
    # elif city_name == u'香港特别行政区':
    #     city_map.append_relational_addr(u'香港', pca_tuple, C)
    # elif city_name == u'澳门特别行政区':
    #     city_map.append_relational_addr(u'澳门', pca_tuple, C)
    # 自治区下的二级区划，eg喀什地区
    elif len(city_name) > 3 and city_name.endswith(u'地区'):
        city_map.append_relational_addr(city_name[:-2], pca_tuple, C)


def _fill_province_map(province_map, record_dict):
    """
    填充一级区划（省级）地名，包括简称
    :param province_map: dict
    :param record_dict: dict
    :return: province_map
    """
    sheng = record_dict[1]  # sheng
    if sheng not in province_map:
        province_map[sheng] = sheng
        # 处理省的简写情况
        # 普通省分 和 直辖市
        if sheng.endswith(u'省') or sheng.endswith(u'市'):
            province_map[sheng[:-1]] = sheng
        # 自治区
        elif sheng == u'新疆维吾尔自治区':
            province_map[u'新疆'] = sheng
        elif sheng == u'内蒙古自治区':
            province_map['内蒙古'] = sheng
        elif sheng == u'广西壮族自治区':
            province_map[u'广西'] = sheng
            province_map[u'广西省'] = sheng
        elif sheng == u'西藏自治区':
            province_map[u'西藏'] = sheng
        elif sheng == u'宁夏回族自治区':
            province_map[u'宁夏'] = sheng
        # 特别行政区
        elif sheng == u'香港特别行政区':
            province_map[u'香港'] = sheng
        elif sheng == u'澳门特别行政区':
            province_map[u'澳门'] = sheng


area_map, city_map, province_area_map, province_map, latlng = _data_from_csv()

# 直辖市
munis = {u'北京市', u'天津市', u'上海市', u'重庆市'}


def is_munis(city_full_name):
    return city_full_name in munis


# 区级到市级的映射
myumap = {
    u'南关区': u'长春市',
    u'南山区': u'深圳市',
    u'宝山区': u'上海市',
    u'普陀区': u'上海市',
    u'浦东区': u'上海市',
    u'市辖区': u'东莞市',
    u'朝阳区': u'北京市',
    u'河东区': u'天津市',
    u'白云区': u'广州市',
    u'西湖区': u'杭州市',
    u'铁西区': u'沈阳市',
}


def transform(location_strs, umap=myumap, index=[], cut=False, lookahead=8, pos_sensitive=False, open_warning=False):
    """将地址描述字符串转换以"省","市","区"信息为列的DataFrame表格
        Args:
            locations:地址描述字符集合,可以是list, Series等任意可以进行for in循环的集合
                      比如:["徐汇区虹漕路461号58号楼5楼", "泉州市洛江区万安塘西工业区"]
            umap:自定义的区级到市级的映射,主要用于解决区重名问题,如果定义的映射在模块中已经存在，则会覆盖模块中自带的映射
            index:可以通过这个参数指定输出的DataFrame的index,默认情况下是range(len(data))
            cut:是否使用分词，默认使用，分词模式速度较快，但是准确率可能会有所下降
            lookahead:只有在cut为false的时候有效，表示最多允许向前看的字符的数量
                      默认值为8是为了能够发现"新疆维吾尔族自治区"这样的长地名
                      如果你的样本中都是短地名的话，可以考虑把这个数字调小一点以提高性能
            pos_sensitive:如果为True则会多返回三列，分别提取出的省市区在字符串中的位置，如果字符串中不存在的话则显示-1
            open_warning: 是否打开umap警告, 默认关闭
        Returns:
            一个Pandas的DataFrame类型的表格，如下：
               |省    |市   |区    |地名                 |
               |上海市|上海市|徐汇区|虹漕路461号58号楼5楼  |
               |福建省|泉州市|洛江区|万安塘西工业区        |
    """
    if not isinstance(location_strs, Iterable):
        from .exceptions import InputTypeNotSuportException
        raise InputTypeNotSuportException(
            'location_strs参数必须为可迭代的类型(比如list, Series等实现了__iter__方法的对象)')

    result = pd.DataFrame(
        [_handle_one_record(addr, umap, cut, lookahead, pos_sensitive, open_warning) for addr in location_strs],
        index=index) \
        if index else pd.DataFrame(
        [_handle_one_record(addr, umap, cut, lookahead, pos_sensitive, open_warning) for addr in location_strs])
    # 这句的唯一作用是让列的顺序好看一些
    if pos_sensitive:
        return result.loc[:, ('省', '市', '区', '地名', '省_pos', '市_pos', '区_pos')]
    else:
        return result.loc[:, ('省', '市', '区', '地名')]


def _handle_one_record(addr, umap, cut, lookahead, pos_sensitive, open_warning):
    """处理一条记录"""
    addr = convert_to_unicode(addr)
    # 空记录
    if not addr or not isinstance(addr, text_type):
        empty = {'省': '', '市': '', '区': ''}
        if pos_sensitive:
            empty['省_pos'] = -1
            empty['市_pos'] = -1
            empty['区_pos'] = -1
        return empty

    # 地名提取
    pca, left_addr = _extract_addr(addr, cut, lookahead)
    # 填充市
    _fill_city(pca, umap, open_warning)
    # 填充省
    _fill_province(pca)

    result = pca.propertys_dict(pos_sensitive)
    result['地名'] = left_addr
    return result


def _fill_province(pca):
    """填充省"""
    if (not pca.province) and pca.city and (pca.city in city_map):
        pca.province = city_map.get_value(pca.city, P)


def _fill_city(pca, umap, open_warning):
    """填充市"""
    if not pca.city:
        # 从 区 映射
        if pca.area:
            # 从umap中映射
            if umap.get(pca.area):
                pca.city = umap.get(pca.area)
                return
            if pca.area in area_map and area_map.is_unique_value(pca.area):
                if pca.province:
                    if area_map.get_value(pca.area, P) == pca.province:
                        pca.city = area_map.get_value(pca.area, C)
                        return
                else:
                    pca.city = area_map.get_value(pca.area, C)
                    return

        # 从 省,区 映射
        if pca.area and pca.province:
            newKey = (pca.province, pca.area)
            if newKey in province_area_map and province_area_map.is_unique_value(newKey):
                pca.city = province_area_map.get_value(newKey, C)
                return

        if open_warning:
            warnings.warn("%s 无法映射, 建议添加进umap中" % pca.area)


def _extract_addr(addr, cut, lookahead):
    """提取地址中的省,市,区名称
       Args:
           addr:原始地址字符串
           cut: 是否分词
       Returns:
           [sheng, shi, qu, (sheng_pos, shi_pos, qu_pos)], addr
    """
    return _jieba_extract(addr) if cut else _full_text_extract(addr, lookahead)


def _jieba_extract(addr):
    """基于结巴分词进行提取"""
    import jieba
    result = Pca()
    pos = 0
    truncate = {0: 0}

    def _set_pca(pca_property, name, full_name):
        """pca_property: 'province', 'city' or 'area'"""
        if not getattr(result, pca_property):
            setattr(result, pca_property, full_name)
            setattr(result, pca_property + "_pos", pos)
            if is_munis(full_name):
                setattr(result, "province_pos", pos)
            # nonlocal truncate, replace with dict
            # refer: https://www.it610.com/article/50433.htm
            if pos == truncate[0]:
                truncate[0] += len(name)

    for word in jieba.cut(addr):
        # 优先提取低级别行政区 (主要是为直辖市和特别行政区考虑)
        if word in area_map:
            _set_pca('area', word, area_map.get_full_name(word))
        elif word in city_map:
            _set_pca('city', word, city_map.get_full_name(word))
        elif word in province_map:
            _set_pca('province', word, province_map[word])

        pos += len(word)

    return result, addr[truncate[0]:]


filter_address_chars = [u'路', u'街', u'村', u'桥']


def _full_text_extract(addr, lookahead):
    """全文匹配进行提取"""
    result = Pca()
    truncate = {0: 0}

    def _set_pca(pca_property, pos, name, full_name):
        """pca_property: 'province', 'city' or 'area'"""

        def _defer_set():
            if not getattr(result, pca_property):
                setattr(result, pca_property, full_name)
                setattr(result, pca_property + "_pos", pos)
                if is_munis(full_name):
                    setattr(result, "province_pos", pos)
                # nonlocal truncate
                if pos == truncate[0]:
                    truncate[0] += len(name)
            return len(name)

        return _defer_set

    # i为起始位置
    i = 0
    while i < len(addr):
        # 用于设置pca属性的函数
        defer_fun = None
        # length为从起始位置开始的长度,从中提取出最长的地址
        for length in range(1, lookahead + 1):
            end_pos = i + length
            if end_pos > len(addr):
                break
            word = addr[i:end_pos]
            word_next = addr[end_pos] if end_pos < len(addr) else ''

            # 优先提取低级别的行政区 (主要是为直辖市和特别行政区考虑)
            if word_next in filter_address_chars:
                continue
            elif word in area_map:
                defer_fun = _set_pca('area', i, word, area_map.get_full_name(word))
                continue
            elif word in city_map:
                defer_fun = _set_pca('city', i, word, city_map.get_full_name(word))
                continue
            elif word in province_map:
                defer_fun = _set_pca('province', i, word, province_map[word])
                continue

        if defer_fun:
            i += defer_fun()
        else:
            i += 1

    return result, addr[truncate[0]:]