feat: 创建 barcode_helper 接口;创建 docker file
All checks were successful
continuous-integration/drone/push Build is passing
continuous-integration/drone Build is passing

This commit is contained in:
Ching 2024-03-02 17:24:30 +08:00
parent 7428f40c46
commit 928d9685f2
6 changed files with 553 additions and 0 deletions

45
.drone.yml Normal file
View File

@ -0,0 +1,45 @@
kind: pipeline
type: docker
name: default
steps:
- name: build-and-push
image: plugins/docker
settings:
username:
from_secret: docker_username
password:
from_secret: docker_password
repo: git.tunpok.com/ching/grocy-barcode-helper
registry: git.tunpok.com
tags: latest
# - name: notify
# image: plugins/webhook
# settings:
# urls: http://bark.tunpok.com/UZ6zC82bKRjQaXiVkosVWh/
# content_type: application/json
# template: |
# {
# "title": "{{#success build.status}}🟢{{else}}🔴{{/success}} Drone Build #{{ build.number }} {{ build.status }} {{#success build.status}}🟢{{else}}🔴{{/success}}",
# "body": "Project: {{ repo.name }}\nBranch: {{ build.branch }}\nCommit: {{ truncate build.commit 8 }}",
# "group": "drone",
# "url": "{{ build.link }}",
# "icon": "https://static-00.iconduck.com/assets.00/drone-icon-2048x2048-6zua2vkz.png"
# }
# when:
# status: [success, failure]
# https://discord.com/api/webhooks/1213409068559900683/szl0AICZwA1V82dg8Vrc_xqOCl1WwnxktlQdf4cdILswR-xZBI5-JOdqGSD8dVUNcUlH
- name: discord notification
image: appleboy/drone-discord
settings:
webhook_id: 1213409068559900683
webhook_token: szl0AICZwA1V82dg8Vrc_xqOCl1WwnxktlQdf4cdILswR-xZBI5-JOdqGSD8dVUNcUlH
when:
status: [success, failure]
volumes:
- name: dockersock
host:
path: /var/run/docker.sock

1
.gitignore vendored
View File

@ -8,6 +8,7 @@ __pycache__/
*.so
# Distribution / packaging
.python-version
.Python
build/
develop-eggs/

14
Dockerfile Normal file
View File

@ -0,0 +1,14 @@
# 使用自定义 Docker Registry 中的官方 Python 镜像作为基础镜像
FROM git.tunpok.com/ching/python-env:latest
# 设置工作目录
WORKDIR /app
# 将当前目录下的所有文件复制到容器中
COPY . .
# port number
EXPOSE 9288
# run flask app
CMD ["python", "app.py"]

218
app.py Normal file
View File

@ -0,0 +1,218 @@
# coding = utf-8
import json
import os
import requests
from flask import Flask, jsonify, request
from pygrocy import EntityType, Grocy
from barcode import BarcodeSpider
# config = configparser.ConfigParser()
# config.read('config.ini')
# GROCY_URL = config.get('Grocy', 'GROCY_URL')
# GROCY_PORT = config.getint('Grocy', 'GROCY_PORT')
# GROCY_API = config.get('Grocy', 'GROCY_API')
# GROCY_DEFAULT_QUANTITY_UNIT_ID = config.getint('Grocy', 'GROCY_DEFAULT_QUANTITY_UNIT_ID')
# GROCY_DEFAULT_BEST_BEFORE_DAYS = config.get('Grocy', 'GROCY_DEFAULT_BEST_BEFORE_DAYS')
# GROCY_LOCATION = {}
# for key in config['GrocyLocation']:
# GROCY_LOCATION[key] = config.get('GrocyLocation', key)
# X_RapidAPI_Key = config.get('RapidAPI', 'X_RapidAPI_Key')
# get config from environment
GROCY_URL = os.environ.get("GROCY_URL")
GROCY_API_KEY = os.environ.get("GROCY_API_KEY")
GROCY_PORT = os.environ.get("GROCY_PORT")
GROCY_DEFAULT_QUANTITY_UNIT_ID = os.environ.get("GROCY_DEFAULT_QUANTITY_UNIT_ID")
GROCY_DEFAULT_BEST_BEFORE_DAYS = os.environ.get("GROCY_DEFAULT_BEST_BEFORE_DAYS")
GROCY_LOCATION = []
X_RapidAPI_Key = os.environ.get("X_RapidAPI_Key")
app = Flask(__name__)
grocy = Grocy(GROCY_URL, GROCY_API_KEY, GROCY_PORT, verify_ssl=True)
def get_locations():
locations = grocy.get_generic_objects_for_type(EntityType.LOCATIONS)
return locations
def add_product(dict_good, location):
good_name = ""
if "description" in dict_good:
good_name = dict_good["description"]
elif "description_cn" in dict_good:
good_name = dict_good["description_cn"]
if not good_name:
return False
location_map = {item['name']: item['id'] for item in GROCY_LOCATION}
data_grocy = {
"name": good_name,
"description": "",
"location_id": location_map[location],
"qu_id_purchase": GROCY_DEFAULT_QUANTITY_UNIT_ID,
"qu_id_stock": GROCY_DEFAULT_QUANTITY_UNIT_ID,
"default_best_before_days": GROCY_DEFAULT_BEST_BEFORE_DAYS,
"default_consume_location_id": location_map[location],
"move_on_open": "1",
}
if ("gpc" in dict_good) and dict_good["gpc"]:
best_before_days = gpc_best_before_days(int(dict_good["gpc"]))
if best_before_days:
data_grocy["default_best_before_days"] = best_before_days
# add product
response_grocy = grocy.add_generic(EntityType.PRODUCTS, data_grocy)
# # add gds info
grocy.set_userfields(
EntityType.PRODUCTS,
int(response_grocy["created_object_id"]),
"GDSInfo",
json.dumps(dict_good, ensure_ascii=False),
)
# add barcode, ex. 06921168593910
data_barcode = {
"product_id": int(response_grocy["created_object_id"]),
"barcode": dict_good["gtin"],
}
grocy.add_generic(EntityType.PRODUCT_BARCODES, data_barcode)
# add barcode, EAN-13, ex. 6921168593910
if dict_good["gtin"].startswith("0"):
data_barcode = {
"product_id": int(response_grocy["created_object_id"]),
"barcode": dict_good["gtin"].strip("0"),
}
grocy.add_generic(EntityType.PRODUCT_BARCODES, data_barcode)
# add picture
pic_url = ""
if ("picfilename" in dict_good) and dict_good["picfilename"]:
pic_url = dict_good["picfilename"]
elif ("picture_filename" in dict_good) and dict_good["picture_filename"]:
pic_url = dict_good["picture_filename"]
if pic_url:
try:
response_img = requests.get(
pic_url,
{
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36"
},
)
if response_img.status_code == 200:
image_data = response_img.content
with open("img.png", "wb") as o:
# output_data = remove(image_data)
o.write(image_data)
grocy.add_product_pic(int(response_grocy["created_object_id"]), "img.png")
except requests.exceptions.RequestException as err:
print("Request error:", err)
grocy.add_product_by_barcode(dict_good["gtin"], 1.0, 0.0)
return True
def gpc_best_before_days(Code):
"""
保质期 类别
7 50370000中类鲜切水果或蔬菜, 50380000中类鲜切水果或蔬菜, 50350000中类未处理或未加工的新鲜叶菜类蔬菜
14 50250000中类未处理或未加工的新鲜水果, 10000025细类牛奶易腐坏, 10006970细类牛奶替代品易腐坏, 10000278细类酸奶易腐坏, 10006979细类酸奶替代品易腐坏
152 50270000中类未处理或未加工的冷冻水果, 50310000中类未处理或未加工的耐储存水果
305 94000000中类粮食作物, 50000000大类食品饮料和烟草, 10120000中类宠物护理用品或食品组合装, 10110000中类宠物食品或饮品
1005 53000000大类美容个人护理和卫生用品, 47100000中类清洁产品, 47190000中类清洁和卫生产品组合装, 51000000大类医疗保健, 10100000中类宠物护理用品
"""
with open("gpc_brick_code.json") as json_file:
gpc_data = json.load(json_file)
best_before_days = {}
best_before_days["7"] = [
50370000,
50380000,
50350000,
]
best_before_days["14"] = [
50250000,
10000025,
10006970,
10000278,
10006979,
]
best_before_days["152"] = [
50270000,
50310000,
]
best_before_days["305"] = [
94000000,
50000000,
10120000,
10110000,
]
best_before_days["670"] = []
best_before_days["1005"] = [
53000000,
47100000,
47190000,
51000000,
10100000,
]
for item in gpc_data["Schema"]:
if item["Code"] == Code:
codes = [item["Code"], item["Code-1"], item["Code-2"], item["Code-3"]]
for day, filter_codes in best_before_days.items():
if any(code in filter_codes for code in codes):
return day
@app.route("/")
def index():
return "Up and running!"
@app.route("/add", methods=["POST"])
def add():
data = request.json
location = data.get("location", "")
barcode = data.get("barcode", "")
GROCY_LOCATION = get_locations()
if not location:
location = GROCY_LOCATION[0]['name']
try:
grocy.product_by_barcode(barcode)
grocy.add_product_by_barcode(barcode, 1.0, 0.0)
response_data = {"message": "Item added successfully"}
return jsonify(response_data), 200
except:
spider = BarcodeSpider(x_rapidapi_key=X_RapidAPI_Key)
good = spider.get_good(barcode)
if add_product(good, location):
response_data = {"message": "New item added successfully"}
return jsonify(response_data), 200
else:
response_data = {"message": "Fail to add new item"}
return jsonify(response_data), 400
@app.route("/consume", methods=["POST"])
def consume():
try:
data = request.json
barcode = data.get("barcode", "")
grocy.consume_product_by_barcode(barcode)
response_data = {"message": "Item removed successfully"}
return jsonify(response_data), 200
except Exception as e:
error_message = str(e)
response_data = {"error": error_message}
return jsonify(response_data), 400
if __name__ == "__main__":
app.run(host="0.0.0.0", port=9288)

261
barcode.py Normal file
View File

@ -0,0 +1,261 @@
# coding = utf-8
import requests
from loguru import logger
import json
class BarcodeSpider:
"""
条形码爬虫类
"""
def __init__(
self,
rapid_api_url="https://barcodes1.p.rapidapi.com/",
x_rapidapi_key="",
x_rapidapi_host="barcodes1.p.rapidapi.com",
):
self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36"
self.base_url = (
"https://bff.gds.org.cn/gds/searching-api/ProductService/homepagestatistic"
)
self.domestic_url = (
"https://bff.gds.org.cn/gds/searching-api/ProductService/ProductListByGTIN"
)
self.domestic_url_simple = "https://bff.gds.org.cn/gds/searching-api/ProductService/ProductSimpleInfoByGTIN"
self.imported_url = "https://bff.gds.org.cn/gds/searching-api/ImportProduct/GetImportProductDataForGtin"
self.imported_url_blk = "https://www.barcodelookup.com/"
self.rapid_api_url = rapid_api_url
self.x_rapidapi_key = x_rapidapi_key
self.x_rapidapi_host = x_rapidapi_host
def get_domestic_good(self, barcode):
session = requests.session()
session.headers.update({"User-Agent": self.user_agent})
response = session.get(self.base_url)
if response.status_code != 200:
logger.error(
"error in getting base_url status_code is {}, barcode is {}".format(
response.status_code, barcode
)
)
return None
payload = {"PageSize": "30", "PageIndex": "1", "SearchItem": str(barcode)}
response_domestic_url = session.get(self.domestic_url, params=payload)
if response_domestic_url.status_code != 200:
logger.error(
"error in getting domestic_url status_code is {}, barcode is {}".format(
response_domestic_url.status_code, barcode
)
)
return None
good = json.loads(response_domestic_url.text)
if good["Code"] == 2:
logger.error("error, {}, barcode is {}".format(good["Msg"], barcode))
return None
if good["Code"] != 1 or good["Data"]["Items"] == []:
logger.error("error, item no found, barcode is {}".format(barcode))
return None
base_id = good["Data"]["Items"][0]["base_id"]
payload = {"gtin": str(barcode), "id": base_id}
response_domestic_url_simple = session.get(
self.domestic_url_simple, params=payload
)
if response_domestic_url_simple.status_code != 200:
return self.rework_good(good["Data"]["Items"][0])
simpleInfo = json.loads(response_domestic_url_simple.text)
if simpleInfo["Code"] != 1:
return self.rework_good(good["Data"]["Items"][0])
if simpleInfo["Data"] != "":
good["Data"]["Items"][0]["simple_info"] = simpleInfo["Data"]
return self.rework_good(good["Data"]["Items"][0])
return self.rework_good(good["Data"]["Items"][0])
def get_imported_good(self, barcode):
session = requests.session()
session.headers.update({"User-Agent": self.user_agent})
response = session.get(self.base_url)
if response.status_code != 200:
logger.error(
"error in getting base_url status_code is {}, barcode is {}".format(
response.status_code, barcode
)
)
good_blk = self.get_imorted_good_from_blk(barcode)
return good_blk
payload = {
"PageSize": "30",
"PageIndex": "1",
"Gtin": str(barcode),
"Description": "",
"AndOr": "0",
}
response_imported_url = session.get(self.imported_url, params=payload)
if response_imported_url.status_code != 200:
logger.error(
"error in getting imported_url status_code is {}, barcode is {}".format(
response_imported_url.status_code, barcode
)
)
good_blk = self.get_imorted_good_from_blk(barcode)
return good_blk
good = json.loads(response_imported_url.text)
if good["Code"] != 1 or good["Data"]["Items"] == []:
logger.error("error, item no found, barcode is {}".format(barcode))
good_blk = self.get_imorted_good_from_blk(barcode)
return good_blk
if (len(good["Data"]["Items"]) == 1) and (
good["Data"]["Items"][0]["description_cn"] is not None
):
return self.rework_good(good["Data"]["Items"][0])
if (len(good["Data"]["Items"]) == 1) and (
good["Data"]["Items"][0]["description_cn"] is None
):
good_blk = self.get_imorted_good_from_blk(barcode)
return good_blk
if len(good["Data"]["Items"]) >= 2:
for item in good["Data"]["Items"]:
if item["realname"] == item["importer_name"]:
return self.rework_good(item)
return self.rework_good(good["Data"]["Items"][0])
def get_imorted_good_from_blk(self, barcode):
if not self.x_rapidapi_key:
return None
good = {}
querystring = {"query": barcode}
headers = {
"X-RapidAPI-Key": self.x_rapidapi_key,
"X-RapidAPI-Host": self.x_rapidapi_host,
}
response = requests.get(self.rapid_api_url, headers=headers, params=querystring)
good_dict = response.json()
if "product" not in good_dict:
return None
good["description_cn"] = good_dict["product"]["title"]
good["picfilename"] = good_dict["product"]["images"][0]
attributes = good_dict["product"]["attributes"]
good["specification_cn"] = ", ".join(
[f"{key}:{value}" for key, value in attributes.items()]
)
good["gtin"] = barcode
return good
def rework_good(self, good):
if "id" in good:
del good["id"]
if "f_id" in good:
del good["f_id"]
if "brandid" in good:
del good["brandid"]
if "base_id" in good:
del good["base_id"]
if good["branch_code"]:
good["branch_code"] = good["branch_code"].strip()
if "picture_filename" in good:
if good["picture_filename"] and (
not good["picture_filename"].startswith("http")
):
good["picture_filename"] = (
"https://oss.gds.org.cn" + good["picture_filename"]
)
if "picfilename" in good:
if good["picfilename"] and (not good["picfilename"].startswith("http")):
good["picfilename"] = "https://oss.gds.org.cn" + good["picfilename"]
return good
def get_good(self, barcode):
if barcode.startswith("69") or barcode.startswith("069"):
return self.get_domestic_good(barcode)
else:
return self.get_imported_good(barcode)
def main():
# 国产商品
# good = BarCodeSpider.get_good('06917878036526')
# 进口商品
# good = BarCodeSpider.get_good('4901201103803')
# 国际商品
good = BarcodeSpider.get_good("3346476426843")
print(good)
if __name__ == "__main__":
main()
"""
国产商品字典
"keyword": "农夫山泉",
"branch_code": "3301 ",
"gtin": "06921168593910",
"specification": "900毫升",
"is_private": false,
"firm_name": "农夫山泉股份有限公司",
"brandcn": "农夫山泉",
"picture_filename": "https://oss.gds.org.cn/userfile/uploada/gra/1712072230/06921168593910/06921168593910.1.jpg",
"description": "农夫山泉NFC橙汁900ml",
"logout_flag": "0",
"have_ms_product": 0,
"base_create_time": "2018-07-10T10:01:31.763Z",
"branch_name": "浙江分中心",
"base_source": "Source",
"gpc": "10000201",
"gpcname": "即饮型调味饮料",
"saledate": "2017-11-30T16:00:00Z",
"saledateyear": 2017,
"base_last_updated": "2019-01-09T02:00:00Z",
"base_user_id": "源数据服务",
"code": "69211685",
"levels": null,
"levels_source": null,
"valid_date": "2023-02-16T16:00:00Z",
"logout_date": null,
"gtinstatus": 1
"""
"""
进口商品字典
"gtin": "04901201103803",
"description_cn": "UCC117速溶综合咖啡90g",
"specification_cn": "90克",
"brand_cn": "悠诗诗",
"gpc": "10000115",
"gpc_name": "速溶咖啡",
"origin_cn": "392",
"origin_name": "日本",
"codeNet": null,
"codeNetContent": null,
"suggested_retail_price": 0,
"suggested_retail_price_unit": "人民币",
"txtKeyword": null,
"picfilename": "https://oss.gds.org.cn/userfile/importcpfile/201911301903478446204015916.png",
"realname": "磨禾(厦门)进出口有限公司",
"branch_code": "3501",
"branch_name": "福建分中心",
"importer_name": "磨禾(厦门)进出口有限公司",
"certificatefilename": null,
"certificatestatus": 0,
"isprivary": 0,
"isconfidentiality": 0,
"datasource": 0
"""
"""
国际商品字典
"""

14
docker-compose.yml Normal file
View File

@ -0,0 +1,14 @@
version: '3'
services:
barcode-helper:
image: git.tunpok.com/ching/grocy-barcode-helper:latest
container_name: barcode-helper
restart: always
environment:
- GROCY_API_KEY=your-api-key
- GROCY_BASE_URL=https://grocy.tunpok.com
- GROCY_BARCODE_HELPER_PORT=443
- GROCY_DEFAULT_QUANTITY_UNIT_ID=1
- GROCY_DEFAULT_BEST_BEFORE_DAYS=365
ports:
- 9288:9288