The following code builds a rudimentary acupuncture database by collecting data from the web.

I would like to hear suggestions about improvements to the database structure, code organization, web-scraping techniques etc.

(NOTE: Some tables created by the db script have not been fully populated. They are left here to inform the reviewer about my overall design strategy, and these would definitely be worked on, revised and expanded further after gathering feedback from the community.)

The block of code commented out under the get_aliases function is dysfunctional, possibly because the owner of the Yibian website disallows direct calls to hrefs ending with .?ano= which lead to webpages containing detailed acupoint data. This results in errors when the target DOM tag cannot be found.

The original intention of that block of code was to build a list of International Code IDs corresponding to each alias point so that these can be easily imported into the acuAliases table, which makes use of this ID as foreign key to link back to the main Acupoint table.

The workaround for the current code under review is to utilize a block of code under the update_acu_alias_table function to half-manually update the null datasets by comparing alias labels (or "headings") and names in their original or transliterated form with the acupoint names in traditional Chinese (acuName_zh) to see if there is a match.

The above strategy works 99.9% of the time, leaving just one special case, TE18, which has different characters and pronunciations for all its possible names and aliases. This, unfortunately, has to be updated manually with the code below the auto-compare-and-update code block.


import sqlite3 as sql
import wikipedia as wp
import pandas as pd
from copy import deepcopy
from typing import List, Set, Dict, Tuple, Type, BinaryIO
from requests import Session, get
import re
from opencc import OpenCC
from pypinyin import pinyin, Style
from itertools import chain
from lookup import get_id
from bs4 import BeautifulSoup
from urllib.parse import urljoin

SCRIPT = '''
    ID TEXT PRIMARY KEY, -- International Standard Code.
    prcID TEXT UNIQUE, -- PRC Standard Code.
    acuName_zh TEXT, 
    acuName_zh_sim TEXT,
    acuName_en TEXT, 
    acuName_tr TEXT, -- transliterated Chinese text
    meridianID TEXT, -- must create column before adding it as a foreign key. 
    FOREIGN KEY (meridianID) REFERENCES Meridian (ID));
    CREATE TABLE IF NOT EXISTS acuLoc ( -- Location of Acupoints.
    acuLoc_desc TEXT, 
    -- acuLoc_desc_en TEXT,
    acuLoc_pos TEXT, -- General position of acupoint: head, upper/lower limb etc.
    FOREIGN KEY (acuID) REFERENCES Acupoint (ID)); -- primary key is also a foreign key.
    CREATE TABLE IF NOT EXISTS acuFind ( -- How to find an Acupoint.
    acuFind_desc TEXT,
    ref TEXT);
    CREATE TABLE IF NOT EXISTS acuEx ( -- Acupoints shared by the extraordinary meridians.
    ID TEXT PRIMARY KEY, -- in the form of TV1, BV2... etc.
    bypass TEXT, -- use primary key of Acupoint table as foreign key.
    meridianID TEXT, 
    FOREIGN KEY (meridianID) REFERENCES Meridian (ID),
    FOREIGN KEY (bypass) REFERENCES Acupoint (ID));
    acuID TEXT, 
    aliasName TEXT,
    aliasSrc TEXT, -- source of aliasName.
    PRIMARY KEY (acuID, aliasName));
    meridianName_zh TEXT, 
    meridianName_zh_sim TEXT,
    meridianName_tr TEXT, -- transliteration; Wikipedia provides values only for extraordinary meridians. 
    meridianName_en TEXT, -- name of organ as meridian name. 
    meridianExtra BOOLEAN DEFAULT "0" NOT NULL CHECK (meridianExtra IN (0, 1)));  -- True for extraordinary meridian
    CREATE TABLE IF NOT EXISTS meridianRoute ( -- 循經路線. Routes taken by the extraordinary meridians are related to the AcuEx table. 
    meridianID TEXT PRIMARY KEY,
    route TEXT, -- route of meridian
    route_src TEXT, -- source of route description
    route_classic TEXT, -- route of meridian; quote 《黃帝內經·靈樞》.
    meridian_img TEXT,
    FOREIGN KEY (meridianID) REFERENCES Meridian (ID));
    CREATE TABLE IF NOT EXISTS Images ( -- Images for archival and display purposes
    ID TEXT, 
    category TEXT, -- Use names of existing tables to define the data category. eg. acuLoc to indicate location of a point. 
    source TEXT, 
    img BLOB,
    PRIMARY KEY (ID, category, source));
    CREATE TABLE IF NOT EXISTS imgLink ( -- Link image to datapoint. 
    imgID TEXT, -- references Images.ID
    refID TEXT, -- references acuID, meridianID etc.; to be derived from imgID
    imgCAT TEXT, -- references Images.category
    imgSRC TEXT, -- references Images.source
    img_desc TEXT, -- description of image in context, if any.
    FOREIGN KEY (imgCAT) REFERENCES Images (category),
    FOREIGN KEY (imgSRC) REFERENCES Images (source),
    FOREIGN KEY (refID) REFERENCES acuLoc (acuID),
    FOREIGN KEY (refID) REFERENCES meridianRoute (meridianID));

def created_tables() -> List[str]:
    r = re.compile("CREATE TABLE IF NOT EXISTS (\\w+)")
    tables = r.findall(SCRIPT)

    return tables

def initialize_database(connect: Type[sql.Connection]):
    """Build the basic database structure from scratch. """

    print("Building the Acupuncture database...")

    # Build database structure
    c = connect.cursor()  # use Cursor object to perform SQL commands

    for table in created_tables():
        # Drop tables if exist
        DROP TABLE IF EXISTS {table};

    # Create tables
    # NOTE: Derived values (such as Yinyang attributes of meridians) will not be stored in the database.

# Furnish data

def get_basic_data(connect: Type[sql.Connection]) -> None:
    """Scrape data from English wikipedia \
    to populate the Acupoint and Meridian tables."""

    print("Furnishing database with data from Wikipedia... ")

    html = wp.page("List_of_acupuncture_points").html()
    df = pd.read_html(html)  # parses all tables into dataframes

    # Set meridian, extraordinary meridian and acupoint data

    meridian = df[0][['Code', 'Chinese Name', 'English']]
    meridian.columns = ['ID', 'meridianName_zh', 'meridianName_en']

    extraordinary_meridian = df[1][['Code', 'Name', 'Transliteration', 'English']]
    extraordinary_meridian.columns = ['ID', 'meridianName_zh', 'meridianName_tr', 'meridianName_en']

    acupoint = pd.concat(df[2:16])[['Point', 'Name', 'Transliteration', 'English']]  # standard :16, all (include 奇穴):18
    acupoint.columns = ['ID', 'acuName_zh', 'acuName_tr', 'acuName_en']  # ID = International Standard Code.


    # Meridian data
    meridian = deepcopy(meridian)  # make sure df is a copy, not a view to avoid SettingwithCopyWarning.
    meridian_list = list(meridian["meridianName_zh"])
    meridian_list_abbrev = [re.search(".+[陰陽明](.+經)", item).group(1) for item in meridian_list]

    extraordinary_meridian = deepcopy(extraordinary_meridian)
    as_list = extraordinary_meridian['meridianName_zh'].tolist()
    split_list = [item.split('; ') for item in as_list]
    sim_list = [sim for sim, zh in split_list]
    zh_list = [zh for sim, zh in split_list]

    extraordinary_meridian['meridianName_zh'] = zh_list
    extraordinary_meridian['meridianName_zh_sim'] = sim_list
    extraordinary_meridian['meridianExtra'] = 1

    # Acupoint data

    acupoint = deepcopy(acupoint)  # make sure df is a copy, not a view.
    name_list = list(acupoint['acuName_zh'])
    acu_list = [re.search("([\u4e00-\u9fff]+)([a-z\\d \\[;()\\]]+)?", item).group(1) for item in name_list]
    acupoint['acuName_zh'] = acu_list  # remove aliases from Chinese name.

    as_list = list(acupoint['ID'])
    split_list = [item.split('-') for item in as_list]
    tag_list = [tag.upper() for tag, sn in split_list]
    sn_list = [sn for tag, sn in split_list]  # serial-number of acupoint index

    tag_list = ["LR" if tag == "LIV" else tag for tag in tag_list]
    tag_list = ["GV" if tag == "DU" else tag for tag in tag_list]
    tag_list = ["CV" if tag == "REN" else tag for tag in tag_list]

    new_id_list = [tag + sn for tag, sn in zip(tag_list, sn_list)]
    meridian_id_list = tag_list
    acupoint["ID"] = new_id_list
    acupoint["meridianID"] = meridian_id_list
    acupoint["acuName_tr"] = [transliterate(item) for item in acupoint["acuName_zh"]]

    cc = OpenCC('t2s')
    acupoint["acuName_zh_sim"] = [cc.convert(item) for item in acu_list]
    meridian["meridianName_zh_sim"] = [cc.convert(item) for item in meridian_list]
    meridian["meridianName_tr"] = [transliterate(item) for item in meridian_list_abbrev]
    meridian["ID"] = ["LR" if tag == "LV" else tag for tag in meridian["ID"]]

    # aliases

    c = connect.cursor()
    for i, item in enumerate(name_list):
        alias = re.search(".+([a-z0-9 \\[;\\(\\)\\]]+)([\u4e00-\u9fff]+)", item)
        if alias:
            INSERT INTO acuAlias (acuID, aliasName, aliasSrc)
                VALUES ("{new_id_list[i]}", "{alias.group(2)}", "wiki");

    # NOTE: This wiki page uses a different code (against the PRC standard) for certain acupoints.
    # e.g. TE instead of SJ for 三焦經; GV instead of DU for 督脈 etc.
    # We'll be using the International Standard for this App.
    # The PRC Standard will be referenced under the Acupoint.prcID column.

    # TE -> SJ (三焦)
    # LV -> LR (肝)
    # CV -> RN
    # GV -> DU

    # NOTE: China has switched to International code by 2012.
    # We keep track of that standard to access「A+醫學百科」data.

    prc_tag_list = ["SJ" if tag == "TE" else tag for tag in tag_list]  # 三焦
    prc_tag_list = ["RN" if tag == "CV" else tag for tag in prc_tag_list]  # 任
    prc_tag_list = ["DU" if tag == "GV" else tag for tag in prc_tag_list]  # 督
    prc_id_list = [tag + sn for tag, sn in zip(prc_tag_list, sn_list)]
    acupoint["prcID"] = prc_id_list

    # Write to database

    acupoint.to_sql('Acupoint', connect, if_exists='append', index=False)  # save to sql database
    meridian.to_sql('Meridian', connect, if_exists='append', index=False)
    extraordinary_meridian.to_sql('Meridian', connect, if_exists='append', index=False)

    # Remove duplicated Simplified Chinese from alias table.
    DELETE FROM `acuAlias`
            (SELECT `acuName_zh_sim` FROM `Acupoint`
            WHERE `acuName_zh_sim` = `aliasName`);

def get_extraordinary_route_data(connect: Type[sql.Connection]) -> (List[str], Dict[str, str]):
    """Scrape data from Chinese wikipedia \
    to furnish extraordinary meridian route data to database."""

    html = wp.page("腧穴列表").html()
    df = pd.read_html(html)

    print("Getting Extraordinary Meridian data...")

    acu_ex = df[1]  # dataframe of acupoints on the extraordinary meridians.
    acu_ex = deepcopy(acu_ex)
    acu_ex = acu_ex.iloc[2:, :]  # slice off 任脈 and 督脈
    acu_ex_list = acu_ex["穴位名稱及序號"].tolist()

    split_list = [item.split(': ') for item in acu_ex_list]
    route_list = [route for route, points in split_list]
    points_list = [points for route, points in split_list]
    meridian_list = list(acu_ex["國際代碼"])

    acu_ex_dict = {}
    for i, lst in enumerate(points_list):
        points = lst.split(" ")
        split_list = [item.split(".") for item in points]
        bypass = [acuID for acuID, acuName in split_list]
        meridian = meridian_list[i]

        for j, bypass_point in enumerate(bypass):
            acu_ex_dict[f"{meridian}{j + 1}"] = bypass_point

    c = connect.cursor()

    for key in acu_ex_dict.keys():
        INSERT INTO acuEx (ID, bypass, meridianID)
        VALUES ("{key}", "{acu_ex_dict[key]}", "{''.join(i for i in key if not i.isdigit())}");

    for i, item in enumerate(route_list):
        INSERT INTO meridianRoute (meridianID, route, route_src) 
                ("{meridian_list[i]}", "{item}", "https://zh.wikipedia.org/wiki/腧穴列表#奇經八脈"); 

def get_location_data(connect: Type[sql.Connection]) -> None:
    """Scrape data from A+醫學百科 \
    to furnish acupoint location data in Chinese."""

    print("Getting acupoint location data...")

    with get('http://cht.a-hospital.com/w/中华人民共和国国家标准·经穴部位') as resp:
        df = pd.read_html(resp.text)

        acu_loc = pd.concat(df[4:18])
        acu_loc = deepcopy(acu_loc)
        acu_loc.columns = ["prcID", "acuName_zh", "acuName_tr", "acuLoc_desc"]

        acu_loc.to_sql('acuLoc_temp', connect, if_exists='replace', index=False)

        c = connect.cursor()
        INSERT INTO `acuLoc`(`acuID`, `acuLoc_desc`)
        SELECT "ID", "acuLoc_desc" FROM `Acupoint`
        JOIN `acuLoc_temp` ON `Acupoint`.`prcID` = `acuLoc_temp`.`prcID`;
        DROP TABLE acuLoc_temp;

def transliterate(string):
    tr = "".join(list(
            pinyin(string, style=Style.NORMAL)))).capitalize()

    return tr

def get_aliases():
    """Furnish alias names of acupoint from 醫砭 website."""
    BASE_URL = 'http://yibian.hopto.org/acu'
    alias_list = []
    heading_list = []
    url_list = []
    id_list = []

    session = Session()
    for i in range(1, 15):

        with session.get (BASE_URL,
                              'mn': 'jing',
                              'sn': i,
                          ) as resp:

            resp.encoding = resp.apparent_encoding
            soup = BeautifulSoup(resp.content, 'html.parser')
            alias_tags = soup.select(".content_small_label:-soup-contains('別名')")

            for tag in alias_tags:
                anchor = tag.find_previous_siblings('a')[0]
                heading = anchor.text
                href = anchor['href']
                aliases = tag.find_next_sibling().text.split(",")

                print(f"Getting aliases for {heading}...")

                url_list.append(urljoin(BASE_URL, href))

    # print("Building id list...")
    # for url in url_list:
    #     with session.get(url,
    #                      headers={"User-Agent":
    #                                   "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:99.0) Gecko/20100101 Firefox/99.0}",
    #                               }
    #                      ) as resp:
    #         resp.raise_for_status()
    #         resp.encoding = resp.apparent_encoding
    #         soup = BeautifulSoup(resp.content, 'html.parser')
    #         print(url)
    #         tag = soup.select_one('td.content_board table td b').nextSibling
    #         id_str = "".join(filter(str.isalnum, tag))
    #         alias_id = re.sub("[^A-Z0-9]+", "", id_str)
    #         id_list.append(alias_id)
    #         print(alias_id)

    return id_list, heading_list, alias_list

def alias_id_is_null(connect: Type[sql.Connection]):
    """Returns the names of acupoints where the ID is null.
    Meaning: Name given on website does not match existing data on our database."""
    c = connect.cursor()
        SELECT `aliasName` FROM `acuAlias`
        WHERE `acuID` ISNULL;
    return [item[0] for item in c.fetchall()]

def get_column(connect: Type[sql.Connection], column, table) -> List[str]:

    c = connect.cursor()
    SELECT {column} FROM {table}; 

    values = [value[0] for value in c.fetchall()]


    return values

def update_acu_alias_table(connect: Type[sql.Connection]):
    id_list, heading_list, alias_list = get_aliases()

    hdg_idx_dict = {}  # index that points to the corresponding heading for each alias list item
    for i, name in enumerate(heading_list):
        for item in alias_list[i]:
            hdg_idx_dict[item] = name

    c = connect.cursor()
    for i, name in enumerate(heading_list):
        for item in alias_list[i]:
                INSERT INTO acuAlias (acuID, aliasName, aliasSrc)
                VALUES ((SELECT ID FROM Acupoint WHERE acuName_zh = "{name}"),

    # Update null values
    null_list = alias_id_is_null(connect)
    db_acuname = get_column(connect, "acuName_zh", "Acupoint")

    for i, item in enumerate(null_list):
        # null list item match db
        heading = hdg_idx_dict[item]
        null_id = get_id(item)
        if item in db_acuname:
            DELETE FROM acuAlias
            WHERE aliasName = "{item}";
            INSERT INTO acuAlias (acuID, aliasName, aliasSrc)
            VALUES ("{null_id}", "{heading}", "醫砭");

    null_list = alias_id_is_null(connect)
    for item in null_list:
        heading = hdg_idx_dict[item]
        # heading tr match db
        hdg_id = get_id(transliterate(heading))  # use transliterated heading to search
        if hdg_id:
            UPDATE acuAlias
            SET acuID = "{hdg_id}"
            WHERE aliasName = "{item}";
                INSERT INTO acuAlias (acuID, aliasName, aliasSrc)
                VALUES ("{hdg_id}", "{heading}", "醫砭");
            except sql.IntegrityError:
                print(f"Acupoint {heading} is already registered as an alias.")

    # Manual

    UPDATE acuAlias
    SET acuID = "TE18", aliasSrc = "醫砭"
    WHERE aliasName = "資脈";
    INSERT INTO acuAlias (acuID, aliasName, aliasSrc)
    VALUES ("TE18", "瘛脈", "醫砭");

if __name__ == '__main__':
    with sql.connect("acu.db") as conn:  # establish connection to database




1 Answer 1


Your Set, Tuple and BinaryIO imports are unused so delete them.

SCRIPT does not belong in this program. It should be moved to a separate .sql file. But further than that, the whole initialize_database routine should not be included in production code. The setup for a program and its runtime should be well-separated.

Inference of created table names using your created_tables regex is not a good idea. If you want your setup script to be idempotent, include drop table if exists statements inline in your SQL verbatim; don't rely on parsing magic to do this.

connect is not a Type[sql.Connection]; it's a Connection (i.e. an instance, not a type).

It's important for the Mandarin strings in


to be moved to variables with English names for international maintainability.

Some of your type hints are incomplete, for instance transliterate(string: str) -> str. A tool like mypy will let you know about this and other instances of gaps and mismatches.

  • \$\begingroup\$ "It's important for the Mandarin strings in .+[陰陽明](.+經) to be moved to variables with English names for international maintainability." The regex basically abbreviates Traditional Chinese meridian names gotten from the website to a shorter form that better matches its English name. Eg. "手太陰肺經" (literally, Hand-Taiyin-Lung meridian) --> "肺經" (Lung meridian). Variables in the present code are all in English. \$\endgroup\$
    – Sati
    Apr 28, 2022 at 1:47
  • \$\begingroup\$ With regard to the created_tables function, I was finding it hard to synchronize the drop table if exists commands with my constantly changing SCRIPT, often leading to errors that interfere with the debugging process, so I wrote that function to automate the keeping-track process. In retrospect, since the drop table script is there just to help me start clean and avoid key violation errors, I might as well remove the database and rebuild it from scratch. \$\endgroup\$
    – Sati
    Apr 28, 2022 at 2:19

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.