aaiezza/Pipfile

## README.md

      
    Raw
  

              README.md
            
          
    Power Up Rewards | The Partial Pin Code problem

Many grocery stores participate in a sort of rewards give-back program called Power Up Rewards.
The company appears to mainly have a presence on the web to serve customers: https://www.poweruppoints.com.
How it works

At the participating store, maybe depending on what is purchased, Power Up coupons are obtained at the time of purchase. The user is then expected to log into the site, and input the pin codes on the coupons. This redeems the points that the pin codes are good for.
The points can then be redeemed for a variety of other products.
In my experience, one pin code seems to be good for 1,000 Power Up points. And a pin code is redeemed after a certain amount of participating products bas been purchased. In my experience, this is $30 worth of participating products.
One of these participating stores is Kroger.
Here is a description of their participation in Power Up found here.

Get your game on, at Kroger family of stores! Start earning points today when you spend $30 on participating POWER UP products and receive rewards! Rewards Pin Codes will be delivered via printout at checkout during your next in-store visit. You’ll need an account to get started, so be sure to sign in or create one today.

Problem

Recently at Kroger, I received power up coupons, but sadly, I was unable to redeem the codes because the store neglected to refill the ink cartridge on the coupon printer. So I am left with a number of characters missing.
The store itself is unable to reproduce the coupons and Power Up customer service does not seem to be available or interested in helping in this situation.
Idea

While this is not my favorite solution, it certainly may get the job done. Here I provide a script that accepts the coupon code, or as much of it as is available legibly, and produces a number of possible codes.
The app then attempts to redeem the Power Up points using the partial code.
Things to discover


Understanding the coupon code format
It seems to be consistent. I have 4 sample codes at my disposal. More would be nice though.

Narrowing the search space
This would also be nice, but is not necessary. I will assume all alpha-numeric characters to start with.


Proving validity of the code
This cannot be done without also reedeming the points. Thus, it will be necessary for the app to:

Have a means by which to submit codes to Power Up
A means by which to interpret the response as success or failure
A means of authorization to a valid Power Up account that points would be applied to in the event of a successful code entry.


Provided these problems are solvable, a solution to this problem seems plausible.
Pin Code format

Here are some example (used) pin codes:

d1nr-qtc-5zni3
d0dr-k6j-90dtx
cw7z-hel-j5bsz
74ti-0b5-d2bnp
7de0-hsm-dfsl3
iojw-jm3-641hq
iojt-lhw-dg01l
dkai-49x-du354
dkak-54h-cmeph
bjgt-6gp-908a9
qwxi-b37-mpjpp
4nab-6mt-ormvw
zj2d-pco-3nw6e
zicx-wql-z34gl
zj2e-lds-x77qm
tl4t-h6x-zkdsg (did not enter this one in time. Expired 2022-10-07)
0nr1-not-qrzuk (Printed in uppercase! First time I've seen that. Was printed on October 1, 2022)
drkj-90i-uh5zc uppercase again, but when entering the code into the website, even trying to input an uppercase petter results in it being changed to lowercase. Letter case does not appear to matter at all, and seems to be only a cosmetic change to the tickets that are being printed when issuing the pin codes. Which is strange to me, because the 0 and o are now more difficult to distinguish from one another. 🤷‍♂️
zc8t-1r3-f3ra1 uppercase
p4vq-wct-b1srw uppercase
ozov-zxy-j1ay8 uppercase
aamb-eix-hbns6 uppercase
aamb-mea-atzlj uppercase
aamc-r6j-jd8o4 uppercase
1lib-42j-nk832 uppercase
kjho-88y-aaq3c uppercase


Break in season. All points have been reset.

CXOP-NOE-J6YAA
V05R-3Z8-T8TFM

Originally was V05R-3Z8-*8TFM, but through trial and error, I found the missing character to be T


YEQ0-YZ1-H0FDR
YEQ1-THV-VO5CY

It appears that the pin codes abide by the following regular expression:
[0-9a-z]{4}-[0-9a-z]{3}-[0-9a-z]{5}
After October 2022, this is the regular expression being used by the generator, however the input on the web application seems to be case insensitive.
[0-9A-Z]{4}-[0-9A-Z]{3}-[0-9A-Z]{5}

  
## main.py
import os
import requests
import re
import logging
import itertools
from time import sleep
from datetime import datetime


num_worker_threads = 1
margin = 0

# logging.basicConfig(level=logging.DEBUG)
session = requests.Session()
session.mount('https://', requests.adapters.HTTPAdapter(pool_connections=1, pool_maxsize=num_worker_threads + margin, max_retries=2))

# https://stackoverflow.com/questions/20658572/python-requests-print-entire-http-request-raw
def pretty_print_POST(req):
    """
    At this point it is completely built and ready
    to be fired; it is "prepared".

    However pay attention at the formatting used in
    this function because it is programmed to be pretty
    printed and may differ from the actual request.
    """
    print('{}\n{}\r\n{}\r\n\r\n{}'.format(
        '-----------START-----------',
        req.method + ' ' + req.url,
        '\r\n'.join('{}: {}'.format(k, v) for k, v in req.headers.items()),
        req.body,
    ))

def put_bank_transaction(row):
  (txid
    , user_id
    , date
    , merchant
    , desc
    , check_number
    , amount
    , note
    , upstream_bank
    , upstream_bank_transaction
    , upstream_bank_account) = row

  assert user_id != None
  assert date != None
  assert merchant != None
  assert amount != None

  def sanitize(s):
    return s.replace('\'', '').replace('(', '').replace(')','').strip()

  data = {
      'user_id': user_id,
      'desc': sanitize(desc),
      'date': date.isoformat(),
      'amount': amount,
      'merchant': sanitize(merchant),
      # 'memo': note,
      'check_number': check_number,
      'upstream_bank_account': upstream_bank_account, # [0:32]
      'upstream_bank_transaction': upstream_bank_transaction, # [0:32]
      'upstream_bank': upstream_bank,
      'upstream_bank_connection_age': 500000 # this prevents Budget API from running de-dupe for this bank transaction
  }

  data = remove_nones_from_dict(data)

  url = os.getenv('EVERYDOLLAR_ENDPOINT_URL')

  # r = requests.Request(
  #   'PUT',
  #   url,
  #   headers = {
  #     'Content-Type': 'application/x-www-form-urlencoded'
  #   },
  #   data = data
  # )
  # prepared = r.prepare()
  # pretty_print_POST(prepared)
  # print()
  # s = requests.Session()
  # response = s.send(prepared)
  try:
    response = session.put(
      url,
      headers = {
        'Content-Type': 'application/x-www-form-urlencoded'
      },
      data = data
    )
  except requests.exceptions.ConnectionError:
    print('Connection refused')
    return (0, 'Connection refused', txid, upstream_bank_transaction)
  return (response.status_code, response.text, txid, upstream_bank_transaction)


def fetch_rows(cur):
  cur.execute(
      """
      SELECT id
            , user_id
            , date
            , merchant
            , "desc"
            , check_number
            , amount
            , note
            , upstream_bank
            , upstream_bank_transaction
            , upstream_bank_account
      FROM transaction_restore_temp
      WHERE budget_response_status = 403
        AND upstream_bank_account IS NOT NULL
        AND upstream_bank_transaction IS NOT NULL
         --LIMIT 100000
      """
    )
  return cur.rowcount


def update_status(status_code, response_body, txid):
  with PGConnection() as (conn,cur):
    try:
      cur.execute(
        """
        UPDATE transaction_restore_temp
        SET budget_response_status = %s
          , budget_response_body = %s
          , budget_response_at = NOW()
        WHERE id = %s
        """,
        (status_code, response_body, txid)
      )
      assert cur.rowcount == 1
    except (Exception, psycopg2.Error) as error:
      print("DB Error ", error)
      print('Rolling back for ', txid)
      conn.rollback()
    except psycopg2.OperationalError:
      print('DB connection error, retrying in one second')
      sleep(1)
      try:
        cur.execute(
          """
          UPDATE transaction_restore_temp
          SET budget_response_status = %s
            , budget_response_body = %s
            , budget_response_at = NOW()
          WHERE id = %s
          """,
          (status_code, response_body, txid)
        )
        assert cur.rowcount == 1
        print('DB UPDATE retry succeeded')
      except psycopg2.OperationalError:
        print('DB connection error, retry failed')
    finally:
      # print('Committing for ', txid)
      conn.commit()

def worker(q):
    while True:
        row = q.get()
        if row is None:
            break
        (status_code, response_body, txid, upstream_bank_transaction) = put_bank_transaction(row)
        print(status_code, 'from Budget API:', response_body, txid, upstream_bank_transaction, 'queue size:', q.qsize())
        if status_code != 0:
          update_status(status_code, response_body, txid)
        q.task_done()


class PinCode:
  def __init__(self, value):
    assert re.match('[0-9a-z]{4}-[0-9a-z]{3}-[0-9a-z]{5}', value)
    self.value = value

  def __str__(self):
    return self.value

  def __repr__(self):
    return self.value


numbers = '0123456789'
alphanum = numbers + 'abcdefghijklmnopqrstuvwxyz'


class PartialPinCode:
  def __init__(self, value):
    assert re.match('[0-9a-z\\.]{4}-[0-9a-z\\.]{3}-[0-9a-z\\.]{5}', value)
    self.value = value

  def get_possible_pin_codes(self):
    unknown_character_count = self.value.count('.')
    pin_codes = []
    for tup in itertools.product(alphanum, repeat=unknown_character_count):
      pin_pieces = list(tup)
      pin_codes.append(PinCode(''.join([p if p != '.' else pin_pieces.pop(0) for p in self.value])))

    return pin_codes

  def __str__(self):
    return self.value

  def __repr__(self):
    return self.value


def main():
  start_time = datetime.now()

  partial_pin_code_input = input('Partial Pin Code: ')
  # partial_pin_code_input = '1.11-55.-.7777'
  partial_pin_code = PartialPinCode(partial_pin_code_input)
  pin_codes = partial_pin_code.get_possible_pin_codes()
  print(pin_codes)

  # assert os.getenv('ENDPOINT_URL')
  # q = queue.Queue()

  # try:
  #   with PGConnection() as (conn,cur):
  #     fetch_rows(cur)
  #     rows = cur.fetchall()
  #     for row in rows:
  #       q.put(row)
  # except KeyboardInterrupt as error:
  #   print("Stopped by keyboard interrupt ", error)
  # except (Exception, psycopg2.Error) as error:
  #   print("Error ", error)
  #   if conn:
  #     conn.rollback()
  #
  # threads = []
  # for i in range(num_worker_threads):
  #     t = threading.Thread(target=worker, args=(q,))
  #     t.start()
  #     threads.append(t)

  # block until all tasks are done
  # q.join()

  # # stop workers
  # for i in range(num_worker_threads):
  #     q.put(None)
  # for t in threads:
  #     t.join()
  #
  # db_pool.closeall()


  run_time = datetime.now() - start_time
  print('Finished in',run_time)


if __name__ == '__main__':
  main()

## Pipfile
[[source]]
url = "https://pypi.org/simple"
verify_ssl = true
name = "pypi"

[packages]
requests = "*"

[dev-packages]
pytest = "*"
pytest-watch = "*"
pylint = "*"

[requires]
python_version = "3.8"
	import os
	import requests
	import re
	import logging
	import itertools
	from time import sleep
	from datetime import datetime


	num_worker_threads = 1
	margin = 0

	# logging.basicConfig(level=logging.DEBUG)
	session = requests.Session()
	session.mount('https://', requests.adapters.HTTPAdapter(pool_connections=1, pool_maxsize=num_worker_threads + margin, max_retries=2))

	# https://stackoverflow.com/questions/20658572/python-requests-print-entire-http-request-raw
	def pretty_print_POST(req):
	"""
	At this point it is completely built and ready
	to be fired; it is "prepared".

	However pay attention at the formatting used in
	this function because it is programmed to be pretty
	printed and may differ from the actual request.
	"""
	print('{}\n{}\r\n{}\r\n\r\n{}'.format(
	'-----------START-----------',
	req.method + ' ' + req.url,
	'\r\n'.join('{}: {}'.format(k, v) for k, v in req.headers.items()),
	req.body,
	))

	def put_bank_transaction(row):
	(txid
	, user_id
	, date
	, merchant
	, desc
	, check_number
	, amount
	, note
	, upstream_bank
	, upstream_bank_transaction
	, upstream_bank_account) = row

	assert user_id != None
	assert date != None
	assert merchant != None
	assert amount != None

	def sanitize(s):
	return s.replace('\'', '').replace('(', '').replace(')','').strip()

	data = {
	'user_id': user_id,
	'desc': sanitize(desc),
	'date': date.isoformat(),
	'amount': amount,
	'merchant': sanitize(merchant),
	# 'memo': note,
	'check_number': check_number,
	'upstream_bank_account': upstream_bank_account, # [0:32]
	'upstream_bank_transaction': upstream_bank_transaction, # [0:32]
	'upstream_bank': upstream_bank,
	'upstream_bank_connection_age': 500000 # this prevents Budget API from running de-dupe for this bank transaction
	}

	data = remove_nones_from_dict(data)

	url = os.getenv('EVERYDOLLAR_ENDPOINT_URL')

	# r = requests.Request(
	# 'PUT',
	# url,
	# headers = {
	# 'Content-Type': 'application/x-www-form-urlencoded'
	# },
	# data = data
	# )
	# prepared = r.prepare()
	# pretty_print_POST(prepared)
	# print()
	# s = requests.Session()
	# response = s.send(prepared)
	try:
	response = session.put(
	url,
	headers = {
	'Content-Type': 'application/x-www-form-urlencoded'
	},
	data = data
	)
	except requests.exceptions.ConnectionError:
	print('Connection refused')
	return (0, 'Connection refused', txid, upstream_bank_transaction)
	return (response.status_code, response.text, txid, upstream_bank_transaction)


	def fetch_rows(cur):
	cur.execute(
	"""
	SELECT id
	, user_id
	, date
	, merchant
	, "desc"
	, check_number
	, amount
	, note
	, upstream_bank
	, upstream_bank_transaction
	, upstream_bank_account
	FROM transaction_restore_temp
	WHERE budget_response_status = 403
	AND upstream_bank_account IS NOT NULL
	AND upstream_bank_transaction IS NOT NULL
	--LIMIT 100000
	"""
	)
	return cur.rowcount


	def update_status(status_code, response_body, txid):
	with PGConnection() as (conn,cur):
	try:
	cur.execute(
	"""
	UPDATE transaction_restore_temp
	SET budget_response_status = %s
	, budget_response_body = %s
	, budget_response_at = NOW()
	WHERE id = %s
	""",
	(status_code, response_body, txid)
	)
	assert cur.rowcount == 1
	except (Exception, psycopg2.Error) as error:
	print("DB Error ", error)
	print('Rolling back for ', txid)
	conn.rollback()
	except psycopg2.OperationalError:
	print('DB connection error, retrying in one second')
	sleep(1)
	try:
	cur.execute(
	"""
	UPDATE transaction_restore_temp
	SET budget_response_status = %s
	, budget_response_body = %s
	, budget_response_at = NOW()
	WHERE id = %s
	""",
	(status_code, response_body, txid)
	)
	assert cur.rowcount == 1
	print('DB UPDATE retry succeeded')
	except psycopg2.OperationalError:
	print('DB connection error, retry failed')
	finally:
	# print('Committing for ', txid)
	conn.commit()

	def worker(q):
	while True:
	row = q.get()
	if row is None:
	break
	(status_code, response_body, txid, upstream_bank_transaction) = put_bank_transaction(row)
	print(status_code, 'from Budget API:', response_body, txid, upstream_bank_transaction, 'queue size:', q.qsize())
	if status_code != 0:
	update_status(status_code, response_body, txid)
	q.task_done()


	class PinCode:
	def __init__(self, value):
	assert re.match('[0-9a-z]{4}-[0-9a-z]{3}-[0-9a-z]{5}', value)
	self.value = value

	def __str__(self):
	return self.value

	def __repr__(self):
	return self.value


	numbers = '0123456789'
	alphanum = numbers + 'abcdefghijklmnopqrstuvwxyz'


	class PartialPinCode:
	def __init__(self, value):
	assert re.match('[0-9a-z\\.]{4}-[0-9a-z\\.]{3}-[0-9a-z\\.]{5}', value)
	self.value = value

	def get_possible_pin_codes(self):
	unknown_character_count = self.value.count('.')
	pin_codes = []
	for tup in itertools.product(alphanum, repeat=unknown_character_count):
	pin_pieces = list(tup)
	pin_codes.append(PinCode(''.join([p if p != '.' else pin_pieces.pop(0) for p in self.value])))

	return pin_codes

	def __str__(self):
	return self.value

	def __repr__(self):
	return self.value


	def main():
	start_time = datetime.now()

	partial_pin_code_input = input('Partial Pin Code: ')
	# partial_pin_code_input = '1.11-55.-.7777'
	partial_pin_code = PartialPinCode(partial_pin_code_input)
	pin_codes = partial_pin_code.get_possible_pin_codes()
	print(pin_codes)

	# assert os.getenv('ENDPOINT_URL')
	# q = queue.Queue()

	# try:
	# with PGConnection() as (conn,cur):
	# fetch_rows(cur)
	# rows = cur.fetchall()
	# for row in rows:
	# q.put(row)
	# except KeyboardInterrupt as error:
	# print("Stopped by keyboard interrupt ", error)
	# except (Exception, psycopg2.Error) as error:
	# print("Error ", error)
	# if conn:
	# conn.rollback()
	#
	# threads = []
	# for i in range(num_worker_threads):
	# t = threading.Thread(target=worker, args=(q,))
	# t.start()
	# threads.append(t)

	# block until all tasks are done
	# q.join()

	# # stop workers
	# for i in range(num_worker_threads):
	# q.put(None)
	# for t in threads:
	# t.join()
	#
	# db_pool.closeall()



	run_time = datetime.now() - start_time
	print('Finished in',run_time)


	if __name__ == '__main__':
	main()
	[[source]]
	url = "https://pypi.org/simple"
	verify_ssl = true
	name = "pypi"

	[packages]
	requests = "*"

	[dev-packages]
	pytest = "*"
	pytest-watch = "*"
	pylint = "*"

	[requires]
	python_version = "3.8"