303 lines
11 KiB
Python
303 lines
11 KiB
Python
# coding = utf-8
|
||
import json
|
||
import os
|
||
import re
|
||
|
||
import requests
|
||
from flask import Flask, jsonify, request
|
||
from pygrocy import EntityType, Grocy
|
||
|
||
from barcode import BarcodeSpider
|
||
from recipe import get_recipe_from_xiachufang
|
||
import base64
|
||
|
||
# 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")
|
||
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
|
||
if 'specification' in dict_good:
|
||
good_name = good_name + " - " + dict_good['specification']
|
||
|
||
locations = get_locations()
|
||
if not location:
|
||
location = locations[0]['name']
|
||
|
||
location_map = {item['name']: item['id'] for item in locations}
|
||
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
|
||
|
||
|
||
def convert_image_to_base64(image_content):
|
||
base64_image = base64.b64encode(image_content).decode("utf-8")
|
||
return base64_image
|
||
|
||
|
||
@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", "")
|
||
|
||
try:
|
||
product = grocy.product_by_barcode(barcode)
|
||
barcode_ = None
|
||
amount = 1.0
|
||
if product:
|
||
for product_barcode in product.product_barcodes:
|
||
if product_barcode.barcode == barcode:
|
||
barcode_ = product_barcode
|
||
break
|
||
if barcode_:
|
||
amount = barcode_.amount
|
||
grocy.add_product_by_barcode(barcode, amount, 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 not good:
|
||
response_data = {"message": "Item not found"}
|
||
return jsonify(response_data), 400
|
||
try:
|
||
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
|
||
except Exception as e:
|
||
if hasattr(e, "message"):
|
||
error_message = e.message
|
||
else:
|
||
error_message = str(e)
|
||
response_data = {"message": error_message}
|
||
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 = {"message": error_message}
|
||
return jsonify(response_data), 400
|
||
|
||
|
||
@app.route("/add_recipe", methods=["POST"])
|
||
def add_recipe():
|
||
data = request.json
|
||
source = data.get("source", "xiachufang")
|
||
url = data.get("url", "")
|
||
if source != "xiachufang":
|
||
response_data = {"message": "Invalid source"}
|
||
return jsonify(response_data), 400
|
||
if not url:
|
||
response_data = {"message": "Invalid url"}
|
||
return jsonify(response_data), 400
|
||
try:
|
||
# https://www.xiachufang.com/recipe/107141278/?recipe_type=1&page_scene=6
|
||
recipe_number = re.search(r'(\d+)', url).group(1)
|
||
resp = grocy.get_generic_objects_for_type(EntityType.RECIPES, ['name~%s' % recipe_number])
|
||
if resp:
|
||
response_data = {"message": "Recipe already exists"}
|
||
return jsonify(response_data), 400
|
||
xcf_recipe = get_recipe_from_xiachufang(url)
|
||
description = "<a href=\"" + url + "\">" + '来源' + "</a>"
|
||
description += "<br>"
|
||
if xcf_recipe['ingredients']:
|
||
description += "<h2>用料</h2>\n"
|
||
description += "<ul>\n"
|
||
for ingredient in xcf_recipe["ingredients"]:
|
||
description += f'<li>{ingredient}</li>\n'
|
||
description += "</ul>\n"
|
||
if xcf_recipe['steps']:
|
||
description += "<h2>做法</h2>\n"
|
||
for idx, step in enumerate(xcf_recipe["steps"]):
|
||
description += f"<h3>步骤 {idx+1}</h3>\n"
|
||
description += f'<p>{step[0]}</p>\n'
|
||
if len(step) == 1:
|
||
description += "<br>"
|
||
continue
|
||
img = requests.get(step[1])
|
||
img_data = 'data:image/png;base64,' + convert_image_to_base64(img.content)
|
||
description += f'\n<img src=\"{img_data}\">\n\n'
|
||
description += "<br>"
|
||
|
||
data_grocy = {
|
||
"name": xcf_recipe["name"] + " - " + recipe_number,
|
||
"description": description
|
||
}
|
||
grocy.add_generic(EntityType.RECIPES, data_grocy)
|
||
response_data = {"message": "Recipe added successfully"}
|
||
except Exception as e:
|
||
error_message = str(e)
|
||
response_data = {"message": error_message}
|
||
return jsonify(response_data), 400
|
||
return jsonify(response_data), 200
|
||
|
||
|
||
if __name__ == "__main__":
|
||
app.run(host="0.0.0.0", port=9288)
|